source: src/linux/universal/linux-3.18/kernel/trace/ring_buffer.c @ 31885

Last change on this file since 31885 was 31885, checked in by brainslayer, 5 weeks ago

update

File size: 130.2 KB
Line 
1/*
2 * Generic ring buffer
3 *
4 * Copyright (C) 2008 Steven Rostedt <srostedt@redhat.com>
5 */
6#include <linux/ftrace_event.h>
7#include <linux/ring_buffer.h>
8#include <linux/trace_clock.h>
9#include <linux/trace_seq.h>
10#include <linux/spinlock.h>
11#include <linux/irq_work.h>
12#include <linux/debugfs.h>
13#include <linux/uaccess.h>
14#include <linux/hardirq.h>
15#include <linux/kthread.h>      /* for self test */
16#include <linux/kmemcheck.h>
17#include <linux/module.h>
18#include <linux/percpu.h>
19#include <linux/mutex.h>
20#include <linux/delay.h>
21#include <linux/slab.h>
22#include <linux/init.h>
23#include <linux/hash.h>
24#include <linux/list.h>
25#include <linux/cpu.h>
26#include <linux/fs.h>
27
28#include <asm/local.h>
29
30static void update_pages_handler(struct work_struct *work);
31
32/*
33 * The ring buffer header is special. We must manually up keep it.
34 */
35int ring_buffer_print_entry_header(struct trace_seq *s)
36{
37        int ret;
38
39        ret = trace_seq_puts(s, "# compressed entry header\n");
40        ret = trace_seq_puts(s, "\ttype_len    :    5 bits\n");
41        ret = trace_seq_puts(s, "\ttime_delta  :   27 bits\n");
42        ret = trace_seq_puts(s, "\tarray       :   32 bits\n");
43        ret = trace_seq_putc(s, '\n');
44        ret = trace_seq_printf(s, "\tpadding     : type == %d\n",
45                               RINGBUF_TYPE_PADDING);
46        ret = trace_seq_printf(s, "\ttime_extend : type == %d\n",
47                               RINGBUF_TYPE_TIME_EXTEND);
48        ret = trace_seq_printf(s, "\tdata max type_len  == %d\n",
49                               RINGBUF_TYPE_DATA_TYPE_LEN_MAX);
50
51        return ret;
52}
53
54/*
55 * The ring buffer is made up of a list of pages. A separate list of pages is
56 * allocated for each CPU. A writer may only write to a buffer that is
57 * associated with the CPU it is currently executing on.  A reader may read
58 * from any per cpu buffer.
59 *
60 * The reader is special. For each per cpu buffer, the reader has its own
61 * reader page. When a reader has read the entire reader page, this reader
62 * page is swapped with another page in the ring buffer.
63 *
64 * Now, as long as the writer is off the reader page, the reader can do what
65 * ever it wants with that page. The writer will never write to that page
66 * again (as long as it is out of the ring buffer).
67 *
68 * Here's some silly ASCII art.
69 *
70 *   +------+
71 *   |reader|          RING BUFFER
72 *   |page  |
73 *   +------+        +---+   +---+   +---+
74 *                   |   |-->|   |-->|   |
75 *                   +---+   +---+   +---+
76 *                     ^               |
77 *                     |               |
78 *                     +---------------+
79 *
80 *
81 *   +------+
82 *   |reader|          RING BUFFER
83 *   |page  |------------------v
84 *   +------+        +---+   +---+   +---+
85 *                   |   |-->|   |-->|   |
86 *                   +---+   +---+   +---+
87 *                     ^               |
88 *                     |               |
89 *                     +---------------+
90 *
91 *
92 *   +------+
93 *   |reader|          RING BUFFER
94 *   |page  |------------------v
95 *   +------+        +---+   +---+   +---+
96 *      ^            |   |-->|   |-->|   |
97 *      |            +---+   +---+   +---+
98 *      |                              |
99 *      |                              |
100 *      +------------------------------+
101 *
102 *
103 *   +------+
104 *   |buffer|          RING BUFFER
105 *   |page  |------------------v
106 *   +------+        +---+   +---+   +---+
107 *      ^            |   |   |   |-->|   |
108 *      |   New      +---+   +---+   +---+
109 *      |  Reader------^               |
110 *      |   page                       |
111 *      +------------------------------+
112 *
113 *
114 * After we make this swap, the reader can hand this page off to the splice
115 * code and be done with it. It can even allocate a new page if it needs to
116 * and swap that into the ring buffer.
117 *
118 * We will be using cmpxchg soon to make all this lockless.
119 *
120 */
121
122/*
123 * A fast way to enable or disable all ring buffers is to
124 * call tracing_on or tracing_off. Turning off the ring buffers
125 * prevents all ring buffers from being recorded to.
126 * Turning this switch on, makes it OK to write to the
127 * ring buffer, if the ring buffer is enabled itself.
128 *
129 * There's three layers that must be on in order to write
130 * to the ring buffer.
131 *
132 * 1) This global flag must be set.
133 * 2) The ring buffer must be enabled for recording.
134 * 3) The per cpu buffer must be enabled for recording.
135 *
136 * In case of an anomaly, this global flag has a bit set that
137 * will permantly disable all ring buffers.
138 */
139
140/*
141 * Global flag to disable all recording to ring buffers
142 *  This has two bits: ON, DISABLED
143 *
144 *  ON   DISABLED
145 * ---- ----------
146 *   0      0        : ring buffers are off
147 *   1      0        : ring buffers are on
148 *   X      1        : ring buffers are permanently disabled
149 */
150
151enum {
152        RB_BUFFERS_ON_BIT       = 0,
153        RB_BUFFERS_DISABLED_BIT = 1,
154};
155
156enum {
157        RB_BUFFERS_ON           = 1 << RB_BUFFERS_ON_BIT,
158        RB_BUFFERS_DISABLED     = 1 << RB_BUFFERS_DISABLED_BIT,
159};
160
161static unsigned long ring_buffer_flags __read_mostly = RB_BUFFERS_ON;
162
163/* Used for individual buffers (after the counter) */
164#define RB_BUFFER_OFF           (1 << 20)
165
166#define BUF_PAGE_HDR_SIZE offsetof(struct buffer_data_page, data)
167
168/**
169 * tracing_off_permanent - permanently disable ring buffers
170 *
171 * This function, once called, will disable all ring buffers
172 * permanently.
173 */
174void tracing_off_permanent(void)
175{
176        set_bit(RB_BUFFERS_DISABLED_BIT, &ring_buffer_flags);
177}
178
179#define RB_EVNT_HDR_SIZE (offsetof(struct ring_buffer_event, array))
180#define RB_ALIGNMENT            4U
181#define RB_MAX_SMALL_DATA       (RB_ALIGNMENT * RINGBUF_TYPE_DATA_TYPE_LEN_MAX)
182#define RB_EVNT_MIN_SIZE        8U      /* two 32bit words */
183
184#ifndef CONFIG_HAVE_64BIT_ALIGNED_ACCESS
185# define RB_FORCE_8BYTE_ALIGNMENT       0
186# define RB_ARCH_ALIGNMENT              RB_ALIGNMENT
187#else
188# define RB_FORCE_8BYTE_ALIGNMENT       1
189# define RB_ARCH_ALIGNMENT              8U
190#endif
191
192#define RB_ALIGN_DATA           __aligned(RB_ARCH_ALIGNMENT)
193
194/* define RINGBUF_TYPE_DATA for 'case RINGBUF_TYPE_DATA:' */
195#define RINGBUF_TYPE_DATA 0 ... RINGBUF_TYPE_DATA_TYPE_LEN_MAX
196
197enum {
198        RB_LEN_TIME_EXTEND = 8,
199        RB_LEN_TIME_STAMP = 16,
200};
201
202#define skip_time_extend(event) \
203        ((struct ring_buffer_event *)((char *)event + RB_LEN_TIME_EXTEND))
204
205static inline int rb_null_event(struct ring_buffer_event *event)
206{
207        return event->type_len == RINGBUF_TYPE_PADDING && !event->time_delta;
208}
209
210static void rb_event_set_padding(struct ring_buffer_event *event)
211{
212        /* padding has a NULL time_delta */
213        event->type_len = RINGBUF_TYPE_PADDING;
214        event->time_delta = 0;
215}
216
217static unsigned
218rb_event_data_length(struct ring_buffer_event *event)
219{
220        unsigned length;
221
222        if (event->type_len)
223                length = event->type_len * RB_ALIGNMENT;
224        else
225                length = event->array[0];
226        return length + RB_EVNT_HDR_SIZE;
227}
228
229/*
230 * Return the length of the given event. Will return
231 * the length of the time extend if the event is a
232 * time extend.
233 */
234static inline unsigned
235rb_event_length(struct ring_buffer_event *event)
236{
237        switch (event->type_len) {
238        case RINGBUF_TYPE_PADDING:
239                if (rb_null_event(event))
240                        /* undefined */
241                        return -1;
242                return  event->array[0] + RB_EVNT_HDR_SIZE;
243
244        case RINGBUF_TYPE_TIME_EXTEND:
245                return RB_LEN_TIME_EXTEND;
246
247        case RINGBUF_TYPE_TIME_STAMP:
248                return RB_LEN_TIME_STAMP;
249
250        case RINGBUF_TYPE_DATA:
251                return rb_event_data_length(event);
252        default:
253                BUG();
254        }
255        /* not hit */
256        return 0;
257}
258
259/*
260 * Return total length of time extend and data,
261 *   or just the event length for all other events.
262 */
263static inline unsigned
264rb_event_ts_length(struct ring_buffer_event *event)
265{
266        unsigned len = 0;
267
268        if (event->type_len == RINGBUF_TYPE_TIME_EXTEND) {
269                /* time extends include the data event after it */
270                len = RB_LEN_TIME_EXTEND;
271                event = skip_time_extend(event);
272        }
273        return len + rb_event_length(event);
274}
275
276/**
277 * ring_buffer_event_length - return the length of the event
278 * @event: the event to get the length of
279 *
280 * Returns the size of the data load of a data event.
281 * If the event is something other than a data event, it
282 * returns the size of the event itself. With the exception
283 * of a TIME EXTEND, where it still returns the size of the
284 * data load of the data event after it.
285 */
286unsigned ring_buffer_event_length(struct ring_buffer_event *event)
287{
288        unsigned length;
289
290        if (event->type_len == RINGBUF_TYPE_TIME_EXTEND)
291                event = skip_time_extend(event);
292
293        length = rb_event_length(event);
294        if (event->type_len > RINGBUF_TYPE_DATA_TYPE_LEN_MAX)
295                return length;
296        length -= RB_EVNT_HDR_SIZE;
297        if (length > RB_MAX_SMALL_DATA + sizeof(event->array[0]))
298                length -= sizeof(event->array[0]);
299        return length;
300}
301EXPORT_SYMBOL_GPL(ring_buffer_event_length);
302
303/* inline for ring buffer fast paths */
304static void *
305rb_event_data(struct ring_buffer_event *event)
306{
307        if (event->type_len == RINGBUF_TYPE_TIME_EXTEND)
308                event = skip_time_extend(event);
309        BUG_ON(event->type_len > RINGBUF_TYPE_DATA_TYPE_LEN_MAX);
310        /* If length is in len field, then array[0] has the data */
311        if (event->type_len)
312                return (void *)&event->array[0];
313        /* Otherwise length is in array[0] and array[1] has the data */
314        return (void *)&event->array[1];
315}
316
317/**
318 * ring_buffer_event_data - return the data of the event
319 * @event: the event to get the data from
320 */
321void *ring_buffer_event_data(struct ring_buffer_event *event)
322{
323        return rb_event_data(event);
324}
325EXPORT_SYMBOL_GPL(ring_buffer_event_data);
326
327#define for_each_buffer_cpu(buffer, cpu)                \
328        for_each_cpu(cpu, buffer->cpumask)
329
330#define TS_SHIFT        27
331#define TS_MASK         ((1ULL << TS_SHIFT) - 1)
332#define TS_DELTA_TEST   (~TS_MASK)
333
334/* Flag when events were overwritten */
335#define RB_MISSED_EVENTS        (1 << 31)
336/* Missed count stored at end */
337#define RB_MISSED_STORED        (1 << 30)
338
339struct buffer_data_page {
340        u64              time_stamp;    /* page time stamp */
341        local_t          commit;        /* write committed index */
342        unsigned char    data[] RB_ALIGN_DATA;  /* data of buffer page */
343};
344
345/*
346 * Note, the buffer_page list must be first. The buffer pages
347 * are allocated in cache lines, which means that each buffer
348 * page will be at the beginning of a cache line, and thus
349 * the least significant bits will be zero. We use this to
350 * add flags in the list struct pointers, to make the ring buffer
351 * lockless.
352 */
353struct buffer_page {
354        struct list_head list;          /* list of buffer pages */
355        local_t          write;         /* index for next write */
356        unsigned         read;          /* index for next read */
357        local_t          entries;       /* entries on this page */
358        unsigned long    real_end;      /* real end of data */
359        struct buffer_data_page *page;  /* Actual data page */
360};
361
362/*
363 * The buffer page counters, write and entries, must be reset
364 * atomically when crossing page boundaries. To synchronize this
365 * update, two counters are inserted into the number. One is
366 * the actual counter for the write position or count on the page.
367 *
368 * The other is a counter of updaters. Before an update happens
369 * the update partition of the counter is incremented. This will
370 * allow the updater to update the counter atomically.
371 *
372 * The counter is 20 bits, and the state data is 12.
373 */
374#define RB_WRITE_MASK           0xfffff
375#define RB_WRITE_INTCNT         (1 << 20)
376
377static void rb_init_page(struct buffer_data_page *bpage)
378{
379        local_set(&bpage->commit, 0);
380}
381
382/**
383 * ring_buffer_page_len - the size of data on the page.
384 * @page: The page to read
385 *
386 * Returns the amount of data on the page, including buffer page header.
387 */
388size_t ring_buffer_page_len(void *page)
389{
390        return local_read(&((struct buffer_data_page *)page)->commit)
391                + BUF_PAGE_HDR_SIZE;
392}
393
394/*
395 * Also stolen from mm/slob.c. Thanks to Mathieu Desnoyers for pointing
396 * this issue out.
397 */
398static void free_buffer_page(struct buffer_page *bpage)
399{
400        free_page((unsigned long)bpage->page);
401        kfree(bpage);
402}
403
404/*
405 * We need to fit the time_stamp delta into 27 bits.
406 */
407static inline int test_time_stamp(u64 delta)
408{
409        if (delta & TS_DELTA_TEST)
410                return 1;
411        return 0;
412}
413
414#define BUF_PAGE_SIZE (PAGE_SIZE - BUF_PAGE_HDR_SIZE)
415
416/* Max payload is BUF_PAGE_SIZE - header (8bytes) */
417#define BUF_MAX_DATA_SIZE (BUF_PAGE_SIZE - (sizeof(u32) * 2))
418
419int ring_buffer_print_page_header(struct trace_seq *s)
420{
421        struct buffer_data_page field;
422        int ret;
423
424        ret = trace_seq_printf(s, "\tfield: u64 timestamp;\t"
425                               "offset:0;\tsize:%u;\tsigned:%u;\n",
426                               (unsigned int)sizeof(field.time_stamp),
427                               (unsigned int)is_signed_type(u64));
428
429        ret = trace_seq_printf(s, "\tfield: local_t commit;\t"
430                               "offset:%u;\tsize:%u;\tsigned:%u;\n",
431                               (unsigned int)offsetof(typeof(field), commit),
432                               (unsigned int)sizeof(field.commit),
433                               (unsigned int)is_signed_type(long));
434
435        ret = trace_seq_printf(s, "\tfield: int overwrite;\t"
436                               "offset:%u;\tsize:%u;\tsigned:%u;\n",
437                               (unsigned int)offsetof(typeof(field), commit),
438                               1,
439                               (unsigned int)is_signed_type(long));
440
441        ret = trace_seq_printf(s, "\tfield: char data;\t"
442                               "offset:%u;\tsize:%u;\tsigned:%u;\n",
443                               (unsigned int)offsetof(typeof(field), data),
444                               (unsigned int)BUF_PAGE_SIZE,
445                               (unsigned int)is_signed_type(char));
446
447        return ret;
448}
449
450struct rb_irq_work {
451        struct irq_work                 work;
452        wait_queue_head_t               waiters;
453        wait_queue_head_t               full_waiters;
454        bool                            waiters_pending;
455        bool                            full_waiters_pending;
456        bool                            wakeup_full;
457};
458
459/*
460 * head_page == tail_page && head == tail then buffer is empty.
461 */
462struct ring_buffer_per_cpu {
463        int                             cpu;
464        atomic_t                        record_disabled;
465        struct ring_buffer              *buffer;
466        raw_spinlock_t                  reader_lock;    /* serialize readers */
467        arch_spinlock_t                 lock;
468        struct lock_class_key           lock_key;
469        unsigned long                   nr_pages;
470        unsigned int                    current_context;
471        struct list_head                *pages;
472        struct buffer_page              *head_page;     /* read from head */
473        struct buffer_page              *tail_page;     /* write to tail */
474        struct buffer_page              *commit_page;   /* committed pages */
475        struct buffer_page              *reader_page;
476        unsigned long                   lost_events;
477        unsigned long                   last_overrun;
478        local_t                         entries_bytes;
479        local_t                         entries;
480        local_t                         overrun;
481        local_t                         commit_overrun;
482        local_t                         dropped_events;
483        local_t                         committing;
484        local_t                         commits;
485        unsigned long                   read;
486        unsigned long                   read_bytes;
487        u64                             write_stamp;
488        u64                             read_stamp;
489        /* ring buffer pages to update, > 0 to add, < 0 to remove */
490        long                            nr_pages_to_update;
491        struct list_head                new_pages; /* new pages to add */
492        struct work_struct              update_pages_work;
493        struct completion               update_done;
494
495        struct rb_irq_work              irq_work;
496};
497
498struct ring_buffer {
499        unsigned                        flags;
500        int                             cpus;
501        atomic_t                        record_disabled;
502        atomic_t                        resize_disabled;
503        cpumask_var_t                   cpumask;
504
505        struct lock_class_key           *reader_lock_key;
506
507        struct mutex                    mutex;
508
509        struct ring_buffer_per_cpu      **buffers;
510
511#ifdef CONFIG_HOTPLUG_CPU
512        struct notifier_block           cpu_notify;
513#endif
514        u64                             (*clock)(void);
515
516        struct rb_irq_work              irq_work;
517};
518
519struct ring_buffer_iter {
520        struct ring_buffer_per_cpu      *cpu_buffer;
521        unsigned long                   head;
522        struct buffer_page              *head_page;
523        struct buffer_page              *cache_reader_page;
524        unsigned long                   cache_read;
525        u64                             read_stamp;
526};
527
528/*
529 * rb_wake_up_waiters - wake up tasks waiting for ring buffer input
530 *
531 * Schedules a delayed work to wake up any task that is blocked on the
532 * ring buffer waiters queue.
533 */
534static void rb_wake_up_waiters(struct irq_work *work)
535{
536        struct rb_irq_work *rbwork = container_of(work, struct rb_irq_work, work);
537
538        wake_up_all(&rbwork->waiters);
539        if (rbwork->wakeup_full) {
540                rbwork->wakeup_full = false;
541                wake_up_all(&rbwork->full_waiters);
542        }
543}
544
545/**
546 * ring_buffer_wait - wait for input to the ring buffer
547 * @buffer: buffer to wait on
548 * @cpu: the cpu buffer to wait on
549 * @full: wait until a full page is available, if @cpu != RING_BUFFER_ALL_CPUS
550 *
551 * If @cpu == RING_BUFFER_ALL_CPUS then the task will wake up as soon
552 * as data is added to any of the @buffer's cpu buffers. Otherwise
553 * it will wait for data to be added to a specific cpu buffer.
554 */
555int ring_buffer_wait(struct ring_buffer *buffer, int cpu, bool full)
556{
557        struct ring_buffer_per_cpu *uninitialized_var(cpu_buffer);
558        DEFINE_WAIT(wait);
559        struct rb_irq_work *work;
560        int ret = 0;
561
562        /*
563         * Depending on what the caller is waiting for, either any
564         * data in any cpu buffer, or a specific buffer, put the
565         * caller on the appropriate wait queue.
566         */
567        if (cpu == RING_BUFFER_ALL_CPUS) {
568                work = &buffer->irq_work;
569                /* Full only makes sense on per cpu reads */
570                full = false;
571        } else {
572                if (!cpumask_test_cpu(cpu, buffer->cpumask))
573                        return -ENODEV;
574                cpu_buffer = buffer->buffers[cpu];
575                work = &cpu_buffer->irq_work;
576        }
577
578
579        while (true) {
580                if (full)
581                        prepare_to_wait(&work->full_waiters, &wait, TASK_INTERRUPTIBLE);
582                else
583                        prepare_to_wait(&work->waiters, &wait, TASK_INTERRUPTIBLE);
584
585                /*
586                 * The events can happen in critical sections where
587                 * checking a work queue can cause deadlocks.
588                 * After adding a task to the queue, this flag is set
589                 * only to notify events to try to wake up the queue
590                 * using irq_work.
591                 *
592                 * We don't clear it even if the buffer is no longer
593                 * empty. The flag only causes the next event to run
594                 * irq_work to do the work queue wake up. The worse
595                 * that can happen if we race with !trace_empty() is that
596                 * an event will cause an irq_work to try to wake up
597                 * an empty queue.
598                 *
599                 * There's no reason to protect this flag either, as
600                 * the work queue and irq_work logic will do the necessary
601                 * synchronization for the wake ups. The only thing
602                 * that is necessary is that the wake up happens after
603                 * a task has been queued. It's OK for spurious wake ups.
604                 */
605                if (full)
606                        work->full_waiters_pending = true;
607                else
608                        work->waiters_pending = true;
609
610                if (signal_pending(current)) {
611                        ret = -EINTR;
612                        break;
613                }
614
615                if (cpu == RING_BUFFER_ALL_CPUS && !ring_buffer_empty(buffer))
616                        break;
617
618                if (cpu != RING_BUFFER_ALL_CPUS &&
619                    !ring_buffer_empty_cpu(buffer, cpu)) {
620                        unsigned long flags;
621                        bool pagebusy;
622
623                        if (!full)
624                                break;
625
626                        raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
627                        pagebusy = cpu_buffer->reader_page == cpu_buffer->commit_page;
628                        raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
629
630                        if (!pagebusy)
631                                break;
632                }
633
634                schedule();
635        }
636
637        if (full)
638                finish_wait(&work->full_waiters, &wait);
639        else
640                finish_wait(&work->waiters, &wait);
641
642        return ret;
643}
644
645/**
646 * ring_buffer_poll_wait - poll on buffer input
647 * @buffer: buffer to wait on
648 * @cpu: the cpu buffer to wait on
649 * @filp: the file descriptor
650 * @poll_table: The poll descriptor
651 *
652 * If @cpu == RING_BUFFER_ALL_CPUS then the task will wake up as soon
653 * as data is added to any of the @buffer's cpu buffers. Otherwise
654 * it will wait for data to be added to a specific cpu buffer.
655 *
656 * Returns POLLIN | POLLRDNORM if data exists in the buffers,
657 * zero otherwise.
658 */
659int ring_buffer_poll_wait(struct ring_buffer *buffer, int cpu,
660                          struct file *filp, poll_table *poll_table)
661{
662        struct ring_buffer_per_cpu *cpu_buffer;
663        struct rb_irq_work *work;
664
665        if (cpu == RING_BUFFER_ALL_CPUS)
666                work = &buffer->irq_work;
667        else {
668                if (!cpumask_test_cpu(cpu, buffer->cpumask))
669                        return -EINVAL;
670
671                cpu_buffer = buffer->buffers[cpu];
672                work = &cpu_buffer->irq_work;
673        }
674
675        poll_wait(filp, &work->waiters, poll_table);
676        work->waiters_pending = true;
677        /*
678         * There's a tight race between setting the waiters_pending and
679         * checking if the ring buffer is empty.  Once the waiters_pending bit
680         * is set, the next event will wake the task up, but we can get stuck
681         * if there's only a single event in.
682         *
683         * FIXME: Ideally, we need a memory barrier on the writer side as well,
684         * but adding a memory barrier to all events will cause too much of a
685         * performance hit in the fast path.  We only need a memory barrier when
686         * the buffer goes from empty to having content.  But as this race is
687         * extremely small, and it's not a problem if another event comes in, we
688         * will fix it later.
689         */
690        smp_mb();
691
692        if ((cpu == RING_BUFFER_ALL_CPUS && !ring_buffer_empty(buffer)) ||
693            (cpu != RING_BUFFER_ALL_CPUS && !ring_buffer_empty_cpu(buffer, cpu)))
694                return POLLIN | POLLRDNORM;
695        return 0;
696}
697
698/* buffer may be either ring_buffer or ring_buffer_per_cpu */
699#define RB_WARN_ON(b, cond)                                             \
700        ({                                                              \
701                int _____ret = unlikely(cond);                          \
702                if (_____ret) {                                         \
703                        if (__same_type(*(b), struct ring_buffer_per_cpu)) { \
704                                struct ring_buffer_per_cpu *__b =       \
705                                        (void *)b;                      \
706                                atomic_inc(&__b->buffer->record_disabled); \
707                        } else                                          \
708                                atomic_inc(&b->record_disabled);        \
709                        WARN_ON(1);                                     \
710                }                                                       \
711                _____ret;                                               \
712        })
713
714/* Up this if you want to test the TIME_EXTENTS and normalization */
715#define DEBUG_SHIFT 0
716
717static inline u64 rb_time_stamp(struct ring_buffer *buffer)
718{
719        /* shift to debug/test normalization and TIME_EXTENTS */
720        return buffer->clock() << DEBUG_SHIFT;
721}
722
723u64 ring_buffer_time_stamp(struct ring_buffer *buffer, int cpu)
724{
725        u64 time;
726
727        preempt_disable_notrace();
728        time = rb_time_stamp(buffer);
729        preempt_enable_no_resched_notrace();
730
731        return time;
732}
733EXPORT_SYMBOL_GPL(ring_buffer_time_stamp);
734
735void ring_buffer_normalize_time_stamp(struct ring_buffer *buffer,
736                                      int cpu, u64 *ts)
737{
738        /* Just stupid testing the normalize function and deltas */
739        *ts >>= DEBUG_SHIFT;
740}
741EXPORT_SYMBOL_GPL(ring_buffer_normalize_time_stamp);
742
743/*
744 * Making the ring buffer lockless makes things tricky.
745 * Although writes only happen on the CPU that they are on,
746 * and they only need to worry about interrupts. Reads can
747 * happen on any CPU.
748 *
749 * The reader page is always off the ring buffer, but when the
750 * reader finishes with a page, it needs to swap its page with
751 * a new one from the buffer. The reader needs to take from
752 * the head (writes go to the tail). But if a writer is in overwrite
753 * mode and wraps, it must push the head page forward.
754 *
755 * Here lies the problem.
756 *
757 * The reader must be careful to replace only the head page, and
758 * not another one. As described at the top of the file in the
759 * ASCII art, the reader sets its old page to point to the next
760 * page after head. It then sets the page after head to point to
761 * the old reader page. But if the writer moves the head page
762 * during this operation, the reader could end up with the tail.
763 *
764 * We use cmpxchg to help prevent this race. We also do something
765 * special with the page before head. We set the LSB to 1.
766 *
767 * When the writer must push the page forward, it will clear the
768 * bit that points to the head page, move the head, and then set
769 * the bit that points to the new head page.
770 *
771 * We also don't want an interrupt coming in and moving the head
772 * page on another writer. Thus we use the second LSB to catch
773 * that too. Thus:
774 *
775 * head->list->prev->next        bit 1          bit 0
776 *                              -------        -------
777 * Normal page                     0              0
778 * Points to head page             0              1
779 * New head page                   1              0
780 *
781 * Note we can not trust the prev pointer of the head page, because:
782 *
783 * +----+       +-----+        +-----+
784 * |    |------>|  T  |---X--->|  N  |
785 * |    |<------|     |        |     |
786 * +----+       +-----+        +-----+
787 *   ^                           ^ |
788 *   |          +-----+          | |
789 *   +----------|  R  |----------+ |
790 *              |     |<-----------+
791 *              +-----+
792 *
793 * Key:  ---X-->  HEAD flag set in pointer
794 *         T      Tail page
795 *         R      Reader page
796 *         N      Next page
797 *
798 * (see __rb_reserve_next() to see where this happens)
799 *
800 *  What the above shows is that the reader just swapped out
801 *  the reader page with a page in the buffer, but before it
802 *  could make the new header point back to the new page added
803 *  it was preempted by a writer. The writer moved forward onto
804 *  the new page added by the reader and is about to move forward
805 *  again.
806 *
807 *  You can see, it is legitimate for the previous pointer of
808 *  the head (or any page) not to point back to itself. But only
809 *  temporarially.
810 */
811
812#define RB_PAGE_NORMAL          0UL
813#define RB_PAGE_HEAD            1UL
814#define RB_PAGE_UPDATE          2UL
815
816
817#define RB_FLAG_MASK            3UL
818
819/* PAGE_MOVED is not part of the mask */
820#define RB_PAGE_MOVED           4UL
821
822/*
823 * rb_list_head - remove any bit
824 */
825static struct list_head *rb_list_head(struct list_head *list)
826{
827        unsigned long val = (unsigned long)list;
828
829        return (struct list_head *)(val & ~RB_FLAG_MASK);
830}
831
832/*
833 * rb_is_head_page - test if the given page is the head page
834 *
835 * Because the reader may move the head_page pointer, we can
836 * not trust what the head page is (it may be pointing to
837 * the reader page). But if the next page is a header page,
838 * its flags will be non zero.
839 */
840static inline int
841rb_is_head_page(struct ring_buffer_per_cpu *cpu_buffer,
842                struct buffer_page *page, struct list_head *list)
843{
844        unsigned long val;
845
846        val = (unsigned long)list->next;
847
848        if ((val & ~RB_FLAG_MASK) != (unsigned long)&page->list)
849                return RB_PAGE_MOVED;
850
851        return val & RB_FLAG_MASK;
852}
853
854/*
855 * rb_is_reader_page
856 *
857 * The unique thing about the reader page, is that, if the
858 * writer is ever on it, the previous pointer never points
859 * back to the reader page.
860 */
861static int rb_is_reader_page(struct buffer_page *page)
862{
863        struct list_head *list = page->list.prev;
864
865        return rb_list_head(list->next) != &page->list;
866}
867
868/*
869 * rb_set_list_to_head - set a list_head to be pointing to head.
870 */
871static void rb_set_list_to_head(struct ring_buffer_per_cpu *cpu_buffer,
872                                struct list_head *list)
873{
874        unsigned long *ptr;
875
876        ptr = (unsigned long *)&list->next;
877        *ptr |= RB_PAGE_HEAD;
878        *ptr &= ~RB_PAGE_UPDATE;
879}
880
881/*
882 * rb_head_page_activate - sets up head page
883 */
884static void rb_head_page_activate(struct ring_buffer_per_cpu *cpu_buffer)
885{
886        struct buffer_page *head;
887
888        head = cpu_buffer->head_page;
889        if (!head)
890                return;
891
892        /*
893         * Set the previous list pointer to have the HEAD flag.
894         */
895        rb_set_list_to_head(cpu_buffer, head->list.prev);
896}
897
898static void rb_list_head_clear(struct list_head *list)
899{
900        unsigned long *ptr = (unsigned long *)&list->next;
901
902        *ptr &= ~RB_FLAG_MASK;
903}
904
905/*
906 * rb_head_page_dactivate - clears head page ptr (for free list)
907 */
908static void
909rb_head_page_deactivate(struct ring_buffer_per_cpu *cpu_buffer)
910{
911        struct list_head *hd;
912
913        /* Go through the whole list and clear any pointers found. */
914        rb_list_head_clear(cpu_buffer->pages);
915
916        list_for_each(hd, cpu_buffer->pages)
917                rb_list_head_clear(hd);
918}
919
920static int rb_head_page_set(struct ring_buffer_per_cpu *cpu_buffer,
921                            struct buffer_page *head,
922                            struct buffer_page *prev,
923                            int old_flag, int new_flag)
924{
925        struct list_head *list;
926        unsigned long val = (unsigned long)&head->list;
927        unsigned long ret;
928
929        list = &prev->list;
930
931        val &= ~RB_FLAG_MASK;
932
933        ret = cmpxchg((unsigned long *)&list->next,
934                      val | old_flag, val | new_flag);
935
936        /* check if the reader took the page */
937        if ((ret & ~RB_FLAG_MASK) != val)
938                return RB_PAGE_MOVED;
939
940        return ret & RB_FLAG_MASK;
941}
942
943static int rb_head_page_set_update(struct ring_buffer_per_cpu *cpu_buffer,
944                                   struct buffer_page *head,
945                                   struct buffer_page *prev,
946                                   int old_flag)
947{
948        return rb_head_page_set(cpu_buffer, head, prev,
949                                old_flag, RB_PAGE_UPDATE);
950}
951
952static int rb_head_page_set_head(struct ring_buffer_per_cpu *cpu_buffer,
953                                 struct buffer_page *head,
954                                 struct buffer_page *prev,
955                                 int old_flag)
956{
957        return rb_head_page_set(cpu_buffer, head, prev,
958                                old_flag, RB_PAGE_HEAD);
959}
960
961static int rb_head_page_set_normal(struct ring_buffer_per_cpu *cpu_buffer,
962                                   struct buffer_page *head,
963                                   struct buffer_page *prev,
964                                   int old_flag)
965{
966        return rb_head_page_set(cpu_buffer, head, prev,
967                                old_flag, RB_PAGE_NORMAL);
968}
969
970static inline void rb_inc_page(struct ring_buffer_per_cpu *cpu_buffer,
971                               struct buffer_page **bpage)
972{
973        struct list_head *p = rb_list_head((*bpage)->list.next);
974
975        *bpage = list_entry(p, struct buffer_page, list);
976}
977
978static struct buffer_page *
979rb_set_head_page(struct ring_buffer_per_cpu *cpu_buffer)
980{
981        struct buffer_page *head;
982        struct buffer_page *page;
983        struct list_head *list;
984        int i;
985
986        if (RB_WARN_ON(cpu_buffer, !cpu_buffer->head_page))
987                return NULL;
988
989        /* sanity check */
990        list = cpu_buffer->pages;
991        if (RB_WARN_ON(cpu_buffer, rb_list_head(list->prev->next) != list))
992                return NULL;
993
994        page = head = cpu_buffer->head_page;
995        /*
996         * It is possible that the writer moves the header behind
997         * where we started, and we miss in one loop.
998         * A second loop should grab the header, but we'll do
999         * three loops just because I'm paranoid.
1000         */
1001        for (i = 0; i < 3; i++) {
1002                do {
1003                        if (rb_is_head_page(cpu_buffer, page, page->list.prev)) {
1004                                cpu_buffer->head_page = page;
1005                                return page;
1006                        }
1007                        rb_inc_page(cpu_buffer, &page);
1008                } while (page != head);
1009        }
1010
1011        RB_WARN_ON(cpu_buffer, 1);
1012
1013        return NULL;
1014}
1015
1016static int rb_head_page_replace(struct buffer_page *old,
1017                                struct buffer_page *new)
1018{
1019        unsigned long *ptr = (unsigned long *)&old->list.prev->next;
1020        unsigned long val;
1021        unsigned long ret;
1022
1023        val = *ptr & ~RB_FLAG_MASK;
1024        val |= RB_PAGE_HEAD;
1025
1026        ret = cmpxchg(ptr, val, (unsigned long)&new->list);
1027
1028        return ret == val;
1029}
1030
1031/*
1032 * rb_tail_page_update - move the tail page forward
1033 *
1034 * Returns 1 if moved tail page, 0 if someone else did.
1035 */
1036static int rb_tail_page_update(struct ring_buffer_per_cpu *cpu_buffer,
1037                               struct buffer_page *tail_page,
1038                               struct buffer_page *next_page)
1039{
1040        struct buffer_page *old_tail;
1041        unsigned long old_entries;
1042        unsigned long old_write;
1043        int ret = 0;
1044
1045        /*
1046         * The tail page now needs to be moved forward.
1047         *
1048         * We need to reset the tail page, but without messing
1049         * with possible erasing of data brought in by interrupts
1050         * that have moved the tail page and are currently on it.
1051         *
1052         * We add a counter to the write field to denote this.
1053         */
1054        old_write = local_add_return(RB_WRITE_INTCNT, &next_page->write);
1055        old_entries = local_add_return(RB_WRITE_INTCNT, &next_page->entries);
1056
1057        /*
1058         * Just make sure we have seen our old_write and synchronize
1059         * with any interrupts that come in.
1060         */
1061        barrier();
1062
1063        /*
1064         * If the tail page is still the same as what we think
1065         * it is, then it is up to us to update the tail
1066         * pointer.
1067         */
1068        if (tail_page == cpu_buffer->tail_page) {
1069                /* Zero the write counter */
1070                unsigned long val = old_write & ~RB_WRITE_MASK;
1071                unsigned long eval = old_entries & ~RB_WRITE_MASK;
1072
1073                /*
1074                 * This will only succeed if an interrupt did
1075                 * not come in and change it. In which case, we
1076                 * do not want to modify it.
1077                 *
1078                 * We add (void) to let the compiler know that we do not care
1079                 * about the return value of these functions. We use the
1080                 * cmpxchg to only update if an interrupt did not already
1081                 * do it for us. If the cmpxchg fails, we don't care.
1082                 */
1083                (void)local_cmpxchg(&next_page->write, old_write, val);
1084                (void)local_cmpxchg(&next_page->entries, old_entries, eval);
1085
1086                /*
1087                 * No need to worry about races with clearing out the commit.
1088                 * it only can increment when a commit takes place. But that
1089                 * only happens in the outer most nested commit.
1090                 */
1091                local_set(&next_page->page->commit, 0);
1092
1093                old_tail = cmpxchg(&cpu_buffer->tail_page,
1094                                   tail_page, next_page);
1095
1096                if (old_tail == tail_page)
1097                        ret = 1;
1098        }
1099
1100        return ret;
1101}
1102
1103static int rb_check_bpage(struct ring_buffer_per_cpu *cpu_buffer,
1104                          struct buffer_page *bpage)
1105{
1106        unsigned long val = (unsigned long)bpage;
1107
1108        if (RB_WARN_ON(cpu_buffer, val & RB_FLAG_MASK))
1109                return 1;
1110
1111        return 0;
1112}
1113
1114/**
1115 * rb_check_list - make sure a pointer to a list has the last bits zero
1116 */
1117static int rb_check_list(struct ring_buffer_per_cpu *cpu_buffer,
1118                         struct list_head *list)
1119{
1120        if (RB_WARN_ON(cpu_buffer, rb_list_head(list->prev) != list->prev))
1121                return 1;
1122        if (RB_WARN_ON(cpu_buffer, rb_list_head(list->next) != list->next))
1123                return 1;
1124        return 0;
1125}
1126
1127/**
1128 * rb_check_pages - integrity check of buffer pages
1129 * @cpu_buffer: CPU buffer with pages to test
1130 *
1131 * As a safety measure we check to make sure the data pages have not
1132 * been corrupted.
1133 */
1134static int rb_check_pages(struct ring_buffer_per_cpu *cpu_buffer)
1135{
1136        struct list_head *head = cpu_buffer->pages;
1137        struct buffer_page *bpage, *tmp;
1138
1139        /* Reset the head page if it exists */
1140        if (cpu_buffer->head_page)
1141                rb_set_head_page(cpu_buffer);
1142
1143        rb_head_page_deactivate(cpu_buffer);
1144
1145        if (RB_WARN_ON(cpu_buffer, head->next->prev != head))
1146                return -1;
1147        if (RB_WARN_ON(cpu_buffer, head->prev->next != head))
1148                return -1;
1149
1150        if (rb_check_list(cpu_buffer, head))
1151                return -1;
1152
1153        list_for_each_entry_safe(bpage, tmp, head, list) {
1154                if (RB_WARN_ON(cpu_buffer,
1155                               bpage->list.next->prev != &bpage->list))
1156                        return -1;
1157                if (RB_WARN_ON(cpu_buffer,
1158                               bpage->list.prev->next != &bpage->list))
1159                        return -1;
1160                if (rb_check_list(cpu_buffer, &bpage->list))
1161                        return -1;
1162        }
1163
1164        rb_head_page_activate(cpu_buffer);
1165
1166        return 0;
1167}
1168
1169static int __rb_allocate_pages(long nr_pages, struct list_head *pages, int cpu)
1170{
1171        struct buffer_page *bpage, *tmp;
1172        long i;
1173
1174        for (i = 0; i < nr_pages; i++) {
1175                struct page *page;
1176                /*
1177                 * __GFP_NORETRY flag makes sure that the allocation fails
1178                 * gracefully without invoking oom-killer and the system is
1179                 * not destabilized.
1180                 */
1181                bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),
1182                                    GFP_KERNEL | __GFP_NORETRY,
1183                                    cpu_to_node(cpu));
1184                if (!bpage)
1185                        goto free_pages;
1186
1187                list_add(&bpage->list, pages);
1188
1189                page = alloc_pages_node(cpu_to_node(cpu),
1190                                        GFP_KERNEL | __GFP_NORETRY, 0);
1191                if (!page)
1192                        goto free_pages;
1193                bpage->page = page_address(page);
1194                rb_init_page(bpage->page);
1195        }
1196
1197        return 0;
1198
1199free_pages:
1200        list_for_each_entry_safe(bpage, tmp, pages, list) {
1201                list_del_init(&bpage->list);
1202                free_buffer_page(bpage);
1203        }
1204
1205        return -ENOMEM;
1206}
1207
1208static int rb_allocate_pages(struct ring_buffer_per_cpu *cpu_buffer,
1209                             unsigned long nr_pages)
1210{
1211        LIST_HEAD(pages);
1212
1213        WARN_ON(!nr_pages);
1214
1215        if (__rb_allocate_pages(nr_pages, &pages, cpu_buffer->cpu))
1216                return -ENOMEM;
1217
1218        /*
1219         * The ring buffer page list is a circular list that does not
1220         * start and end with a list head. All page list items point to
1221         * other pages.
1222         */
1223        cpu_buffer->pages = pages.next;
1224        list_del(&pages);
1225
1226        cpu_buffer->nr_pages = nr_pages;
1227
1228        rb_check_pages(cpu_buffer);
1229
1230        return 0;
1231}
1232
1233static struct ring_buffer_per_cpu *
1234rb_allocate_cpu_buffer(struct ring_buffer *buffer, long nr_pages, int cpu)
1235{
1236        struct ring_buffer_per_cpu *cpu_buffer;
1237        struct buffer_page *bpage;
1238        struct page *page;
1239        int ret;
1240
1241        cpu_buffer = kzalloc_node(ALIGN(sizeof(*cpu_buffer), cache_line_size()),
1242                                  GFP_KERNEL, cpu_to_node(cpu));
1243        if (!cpu_buffer)
1244                return NULL;
1245
1246        cpu_buffer->cpu = cpu;
1247        cpu_buffer->buffer = buffer;
1248        raw_spin_lock_init(&cpu_buffer->reader_lock);
1249        lockdep_set_class(&cpu_buffer->reader_lock, buffer->reader_lock_key);
1250        cpu_buffer->lock = (arch_spinlock_t)__ARCH_SPIN_LOCK_UNLOCKED;
1251        INIT_WORK(&cpu_buffer->update_pages_work, update_pages_handler);
1252        init_completion(&cpu_buffer->update_done);
1253        init_irq_work(&cpu_buffer->irq_work.work, rb_wake_up_waiters);
1254        init_waitqueue_head(&cpu_buffer->irq_work.waiters);
1255        init_waitqueue_head(&cpu_buffer->irq_work.full_waiters);
1256
1257        bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),
1258                            GFP_KERNEL, cpu_to_node(cpu));
1259        if (!bpage)
1260                goto fail_free_buffer;
1261
1262        rb_check_bpage(cpu_buffer, bpage);
1263
1264        cpu_buffer->reader_page = bpage;
1265        page = alloc_pages_node(cpu_to_node(cpu), GFP_KERNEL, 0);
1266        if (!page)
1267                goto fail_free_reader;
1268        bpage->page = page_address(page);
1269        rb_init_page(bpage->page);
1270
1271        INIT_LIST_HEAD(&cpu_buffer->reader_page->list);
1272        INIT_LIST_HEAD(&cpu_buffer->new_pages);
1273
1274        ret = rb_allocate_pages(cpu_buffer, nr_pages);
1275        if (ret < 0)
1276                goto fail_free_reader;
1277
1278        cpu_buffer->head_page
1279                = list_entry(cpu_buffer->pages, struct buffer_page, list);
1280        cpu_buffer->tail_page = cpu_buffer->commit_page = cpu_buffer->head_page;
1281
1282        rb_head_page_activate(cpu_buffer);
1283
1284        return cpu_buffer;
1285
1286 fail_free_reader:
1287        free_buffer_page(cpu_buffer->reader_page);
1288
1289 fail_free_buffer:
1290        kfree(cpu_buffer);
1291        return NULL;
1292}
1293
1294static void rb_free_cpu_buffer(struct ring_buffer_per_cpu *cpu_buffer)
1295{
1296        struct list_head *head = cpu_buffer->pages;
1297        struct buffer_page *bpage, *tmp;
1298
1299        free_buffer_page(cpu_buffer->reader_page);
1300
1301        rb_head_page_deactivate(cpu_buffer);
1302
1303        if (head) {
1304                list_for_each_entry_safe(bpage, tmp, head, list) {
1305                        list_del_init(&bpage->list);
1306                        free_buffer_page(bpage);
1307                }
1308                bpage = list_entry(head, struct buffer_page, list);
1309                free_buffer_page(bpage);
1310        }
1311
1312        kfree(cpu_buffer);
1313}
1314
1315#ifdef CONFIG_HOTPLUG_CPU
1316static int rb_cpu_notify(struct notifier_block *self,
1317                         unsigned long action, void *hcpu);
1318#endif
1319
1320/**
1321 * __ring_buffer_alloc - allocate a new ring_buffer
1322 * @size: the size in bytes per cpu that is needed.
1323 * @flags: attributes to set for the ring buffer.
1324 *
1325 * Currently the only flag that is available is the RB_FL_OVERWRITE
1326 * flag. This flag means that the buffer will overwrite old data
1327 * when the buffer wraps. If this flag is not set, the buffer will
1328 * drop data when the tail hits the head.
1329 */
1330struct ring_buffer *__ring_buffer_alloc(unsigned long size, unsigned flags,
1331                                        struct lock_class_key *key)
1332{
1333        struct ring_buffer *buffer;
1334        long nr_pages;
1335        int bsize;
1336        int cpu;
1337
1338        /* keep it in its own cache line */
1339        buffer = kzalloc(ALIGN(sizeof(*buffer), cache_line_size()),
1340                         GFP_KERNEL);
1341        if (!buffer)
1342                return NULL;
1343
1344        if (!alloc_cpumask_var(&buffer->cpumask, GFP_KERNEL))
1345                goto fail_free_buffer;
1346
1347        nr_pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
1348        buffer->flags = flags;
1349        buffer->clock = trace_clock_local;
1350        buffer->reader_lock_key = key;
1351
1352        init_irq_work(&buffer->irq_work.work, rb_wake_up_waiters);
1353        init_waitqueue_head(&buffer->irq_work.waiters);
1354
1355        /* need at least two pages */
1356        if (nr_pages < 2)
1357                nr_pages = 2;
1358
1359        /*
1360         * In case of non-hotplug cpu, if the ring-buffer is allocated
1361         * in early initcall, it will not be notified of secondary cpus.
1362         * In that off case, we need to allocate for all possible cpus.
1363         */
1364#ifdef CONFIG_HOTPLUG_CPU
1365        cpu_notifier_register_begin();
1366        cpumask_copy(buffer->cpumask, cpu_online_mask);
1367#else
1368        cpumask_copy(buffer->cpumask, cpu_possible_mask);
1369#endif
1370        buffer->cpus = nr_cpu_ids;
1371
1372        bsize = sizeof(void *) * nr_cpu_ids;
1373        buffer->buffers = kzalloc(ALIGN(bsize, cache_line_size()),
1374                                  GFP_KERNEL);
1375        if (!buffer->buffers)
1376                goto fail_free_cpumask;
1377
1378        for_each_buffer_cpu(buffer, cpu) {
1379                buffer->buffers[cpu] =
1380                        rb_allocate_cpu_buffer(buffer, nr_pages, cpu);
1381                if (!buffer->buffers[cpu])
1382                        goto fail_free_buffers;
1383        }
1384
1385#ifdef CONFIG_HOTPLUG_CPU
1386        buffer->cpu_notify.notifier_call = rb_cpu_notify;
1387        buffer->cpu_notify.priority = 0;
1388        __register_cpu_notifier(&buffer->cpu_notify);
1389        cpu_notifier_register_done();
1390#endif
1391
1392        mutex_init(&buffer->mutex);
1393
1394        return buffer;
1395
1396 fail_free_buffers:
1397        for_each_buffer_cpu(buffer, cpu) {
1398                if (buffer->buffers[cpu])
1399                        rb_free_cpu_buffer(buffer->buffers[cpu]);
1400        }
1401        kfree(buffer->buffers);
1402
1403 fail_free_cpumask:
1404        free_cpumask_var(buffer->cpumask);
1405#ifdef CONFIG_HOTPLUG_CPU
1406        cpu_notifier_register_done();
1407#endif
1408
1409 fail_free_buffer:
1410        kfree(buffer);
1411        return NULL;
1412}
1413EXPORT_SYMBOL_GPL(__ring_buffer_alloc);
1414
1415/**
1416 * ring_buffer_free - free a ring buffer.
1417 * @buffer: the buffer to free.
1418 */
1419void
1420ring_buffer_free(struct ring_buffer *buffer)
1421{
1422        int cpu;
1423
1424#ifdef CONFIG_HOTPLUG_CPU
1425        cpu_notifier_register_begin();
1426        __unregister_cpu_notifier(&buffer->cpu_notify);
1427#endif
1428
1429        for_each_buffer_cpu(buffer, cpu)
1430                rb_free_cpu_buffer(buffer->buffers[cpu]);
1431
1432#ifdef CONFIG_HOTPLUG_CPU
1433        cpu_notifier_register_done();
1434#endif
1435
1436        kfree(buffer->buffers);
1437        free_cpumask_var(buffer->cpumask);
1438
1439        kfree(buffer);
1440}
1441EXPORT_SYMBOL_GPL(ring_buffer_free);
1442
1443void ring_buffer_set_clock(struct ring_buffer *buffer,
1444                           u64 (*clock)(void))
1445{
1446        buffer->clock = clock;
1447}
1448
1449static void rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer);
1450
1451static inline unsigned long rb_page_entries(struct buffer_page *bpage)
1452{
1453        return local_read(&bpage->entries) & RB_WRITE_MASK;
1454}
1455
1456static inline unsigned long rb_page_write(struct buffer_page *bpage)
1457{
1458        return local_read(&bpage->write) & RB_WRITE_MASK;
1459}
1460
1461static int
1462rb_remove_pages(struct ring_buffer_per_cpu *cpu_buffer, unsigned long nr_pages)
1463{
1464        struct list_head *tail_page, *to_remove, *next_page;
1465        struct buffer_page *to_remove_page, *tmp_iter_page;
1466        struct buffer_page *last_page, *first_page;
1467        unsigned long nr_removed;
1468        unsigned long head_bit;
1469        int page_entries;
1470
1471        head_bit = 0;
1472
1473        raw_spin_lock_irq(&cpu_buffer->reader_lock);
1474        atomic_inc(&cpu_buffer->record_disabled);
1475        /*
1476         * We don't race with the readers since we have acquired the reader
1477         * lock. We also don't race with writers after disabling recording.
1478         * This makes it easy to figure out the first and the last page to be
1479         * removed from the list. We unlink all the pages in between including
1480         * the first and last pages. This is done in a busy loop so that we
1481         * lose the least number of traces.
1482         * The pages are freed after we restart recording and unlock readers.
1483         */
1484        tail_page = &cpu_buffer->tail_page->list;
1485
1486        /*
1487         * tail page might be on reader page, we remove the next page
1488         * from the ring buffer
1489         */
1490        if (cpu_buffer->tail_page == cpu_buffer->reader_page)
1491                tail_page = rb_list_head(tail_page->next);
1492        to_remove = tail_page;
1493
1494        /* start of pages to remove */
1495        first_page = list_entry(rb_list_head(to_remove->next),
1496                                struct buffer_page, list);
1497
1498        for (nr_removed = 0; nr_removed < nr_pages; nr_removed++) {
1499                to_remove = rb_list_head(to_remove)->next;
1500                head_bit |= (unsigned long)to_remove & RB_PAGE_HEAD;
1501        }
1502
1503        next_page = rb_list_head(to_remove)->next;
1504
1505        /*
1506         * Now we remove all pages between tail_page and next_page.
1507         * Make sure that we have head_bit value preserved for the
1508         * next page
1509         */
1510        tail_page->next = (struct list_head *)((unsigned long)next_page |
1511                                                head_bit);
1512        next_page = rb_list_head(next_page);
1513        next_page->prev = tail_page;
1514
1515        /* make sure pages points to a valid page in the ring buffer */
1516        cpu_buffer->pages = next_page;
1517
1518        /* update head page */
1519        if (head_bit)
1520                cpu_buffer->head_page = list_entry(next_page,
1521                                                struct buffer_page, list);
1522
1523        /*
1524         * change read pointer to make sure any read iterators reset
1525         * themselves
1526         */
1527        cpu_buffer->read = 0;
1528
1529        /* pages are removed, resume tracing and then free the pages */
1530        atomic_dec(&cpu_buffer->record_disabled);
1531        raw_spin_unlock_irq(&cpu_buffer->reader_lock);
1532
1533        RB_WARN_ON(cpu_buffer, list_empty(cpu_buffer->pages));
1534
1535        /* last buffer page to remove */
1536        last_page = list_entry(rb_list_head(to_remove), struct buffer_page,
1537                                list);
1538        tmp_iter_page = first_page;
1539
1540        do {
1541                to_remove_page = tmp_iter_page;
1542                rb_inc_page(cpu_buffer, &tmp_iter_page);
1543
1544                /* update the counters */
1545                page_entries = rb_page_entries(to_remove_page);
1546                if (page_entries) {
1547                        /*
1548                         * If something was added to this page, it was full
1549                         * since it is not the tail page. So we deduct the
1550                         * bytes consumed in ring buffer from here.
1551                         * Increment overrun to account for the lost events.
1552                         */
1553                        local_add(page_entries, &cpu_buffer->overrun);
1554                        local_sub(BUF_PAGE_SIZE, &cpu_buffer->entries_bytes);
1555                }
1556
1557                /*
1558                 * We have already removed references to this list item, just
1559                 * free up the buffer_page and its page
1560                 */
1561                free_buffer_page(to_remove_page);
1562                nr_removed--;
1563
1564        } while (to_remove_page != last_page);
1565
1566        RB_WARN_ON(cpu_buffer, nr_removed);
1567
1568        return nr_removed == 0;
1569}
1570
1571static int
1572rb_insert_pages(struct ring_buffer_per_cpu *cpu_buffer)
1573{
1574        struct list_head *pages = &cpu_buffer->new_pages;
1575        int retries, success;
1576
1577        raw_spin_lock_irq(&cpu_buffer->reader_lock);
1578        /*
1579         * We are holding the reader lock, so the reader page won't be swapped
1580         * in the ring buffer. Now we are racing with the writer trying to
1581         * move head page and the tail page.
1582         * We are going to adapt the reader page update process where:
1583         * 1. We first splice the start and end of list of new pages between
1584         *    the head page and its previous page.
1585         * 2. We cmpxchg the prev_page->next to point from head page to the
1586         *    start of new pages list.
1587         * 3. Finally, we update the head->prev to the end of new list.
1588         *
1589         * We will try this process 10 times, to make sure that we don't keep
1590         * spinning.
1591         */
1592        retries = 10;
1593        success = 0;
1594        while (retries--) {
1595                struct list_head *head_page, *prev_page, *r;
1596                struct list_head *last_page, *first_page;
1597                struct list_head *head_page_with_bit;
1598
1599                head_page = &rb_set_head_page(cpu_buffer)->list;
1600                if (!head_page)
1601                        break;
1602                prev_page = head_page->prev;
1603
1604                first_page = pages->next;
1605                last_page  = pages->prev;
1606
1607                head_page_with_bit = (struct list_head *)
1608                                     ((unsigned long)head_page | RB_PAGE_HEAD);
1609
1610                last_page->next = head_page_with_bit;
1611                first_page->prev = prev_page;
1612
1613                r = cmpxchg(&prev_page->next, head_page_with_bit, first_page);
1614
1615                if (r == head_page_with_bit) {
1616                        /*
1617                         * yay, we replaced the page pointer to our new list,
1618                         * now, we just have to update to head page's prev
1619                         * pointer to point to end of list
1620                         */
1621                        head_page->prev = last_page;
1622                        success = 1;
1623                        break;
1624                }
1625        }
1626
1627        if (success)
1628                INIT_LIST_HEAD(pages);
1629        /*
1630         * If we weren't successful in adding in new pages, warn and stop
1631         * tracing
1632         */
1633        RB_WARN_ON(cpu_buffer, !success);
1634        raw_spin_unlock_irq(&cpu_buffer->reader_lock);
1635
1636        /* free pages if they weren't inserted */
1637        if (!success) {
1638                struct buffer_page *bpage, *tmp;
1639                list_for_each_entry_safe(bpage, tmp, &cpu_buffer->new_pages,
1640                                         list) {
1641                        list_del_init(&bpage->list);
1642                        free_buffer_page(bpage);
1643                }
1644        }
1645        return success;
1646}
1647
1648static void rb_update_pages(struct ring_buffer_per_cpu *cpu_buffer)
1649{
1650        int success;
1651
1652        if (cpu_buffer->nr_pages_to_update > 0)
1653                success = rb_insert_pages(cpu_buffer);
1654        else
1655                success = rb_remove_pages(cpu_buffer,
1656                                        -cpu_buffer->nr_pages_to_update);
1657
1658        if (success)
1659                cpu_buffer->nr_pages += cpu_buffer->nr_pages_to_update;
1660}
1661
1662static void update_pages_handler(struct work_struct *work)
1663{
1664        struct ring_buffer_per_cpu *cpu_buffer = container_of(work,
1665                        struct ring_buffer_per_cpu, update_pages_work);
1666        rb_update_pages(cpu_buffer);
1667        complete(&cpu_buffer->update_done);
1668}
1669
1670/**
1671 * ring_buffer_resize - resize the ring buffer
1672 * @buffer: the buffer to resize.
1673 * @size: the new size.
1674 * @cpu_id: the cpu buffer to resize
1675 *
1676 * Minimum size is 2 * BUF_PAGE_SIZE.
1677 *
1678 * Returns 0 on success and < 0 on failure.
1679 */
1680int ring_buffer_resize(struct ring_buffer *buffer, unsigned long size,
1681                        int cpu_id)
1682{
1683        struct ring_buffer_per_cpu *cpu_buffer;
1684        unsigned long nr_pages;
1685        int cpu, err = 0;
1686
1687        /*
1688         * Always succeed at resizing a non-existent buffer:
1689         */
1690        if (!buffer)
1691                return size;
1692
1693        /* Make sure the requested buffer exists */
1694        if (cpu_id != RING_BUFFER_ALL_CPUS &&
1695            !cpumask_test_cpu(cpu_id, buffer->cpumask))
1696                return size;
1697
1698        nr_pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
1699
1700        /* we need a minimum of two pages */
1701        if (nr_pages < 2)
1702                nr_pages = 2;
1703
1704        size = nr_pages * BUF_PAGE_SIZE;
1705
1706        /*
1707         * Don't succeed if resizing is disabled, as a reader might be
1708         * manipulating the ring buffer and is expecting a sane state while
1709         * this is true.
1710         */
1711        if (atomic_read(&buffer->resize_disabled))
1712                return -EBUSY;
1713
1714        /* prevent another thread from changing buffer sizes */
1715        mutex_lock(&buffer->mutex);
1716
1717        if (cpu_id == RING_BUFFER_ALL_CPUS) {
1718                /* calculate the pages to update */
1719                for_each_buffer_cpu(buffer, cpu) {
1720                        cpu_buffer = buffer->buffers[cpu];
1721
1722                        cpu_buffer->nr_pages_to_update = nr_pages -
1723                                                        cpu_buffer->nr_pages;
1724                        /*
1725                         * nothing more to do for removing pages or no update
1726                         */
1727                        if (cpu_buffer->nr_pages_to_update <= 0)
1728                                continue;
1729                        /*
1730                         * to add pages, make sure all new pages can be
1731                         * allocated without receiving ENOMEM
1732                         */
1733                        INIT_LIST_HEAD(&cpu_buffer->new_pages);
1734                        if (__rb_allocate_pages(cpu_buffer->nr_pages_to_update,
1735                                                &cpu_buffer->new_pages, cpu)) {
1736                                /* not enough memory for new pages */
1737                                err = -ENOMEM;
1738                                goto out_err;
1739                        }
1740                }
1741
1742                get_online_cpus();
1743                /*
1744                 * Fire off all the required work handlers
1745                 * We can't schedule on offline CPUs, but it's not necessary
1746                 * since we can change their buffer sizes without any race.
1747                 */
1748                for_each_buffer_cpu(buffer, cpu) {
1749                        cpu_buffer = buffer->buffers[cpu];
1750                        if (!cpu_buffer->nr_pages_to_update)
1751                                continue;
1752
1753                        /* Can't run something on an offline CPU. */
1754                        if (!cpu_online(cpu)) {
1755                                rb_update_pages(cpu_buffer);
1756                                cpu_buffer->nr_pages_to_update = 0;
1757                        } else {
1758                                schedule_work_on(cpu,
1759                                                &cpu_buffer->update_pages_work);
1760                        }
1761                }
1762
1763                /* wait for all the updates to complete */
1764                for_each_buffer_cpu(buffer, cpu) {
1765                        cpu_buffer = buffer->buffers[cpu];
1766                        if (!cpu_buffer->nr_pages_to_update)
1767                                continue;
1768
1769                        if (cpu_online(cpu))
1770                                wait_for_completion(&cpu_buffer->update_done);
1771                        cpu_buffer->nr_pages_to_update = 0;
1772                }
1773
1774                put_online_cpus();
1775        } else {
1776                /* Make sure this CPU has been intitialized */
1777                if (!cpumask_test_cpu(cpu_id, buffer->cpumask))
1778                        goto out;
1779
1780                cpu_buffer = buffer->buffers[cpu_id];
1781
1782                if (nr_pages == cpu_buffer->nr_pages)
1783                        goto out;
1784
1785                cpu_buffer->nr_pages_to_update = nr_pages -
1786                                                cpu_buffer->nr_pages;
1787
1788                INIT_LIST_HEAD(&cpu_buffer->new_pages);
1789                if (cpu_buffer->nr_pages_to_update > 0 &&
1790                        __rb_allocate_pages(cpu_buffer->nr_pages_to_update,
1791                                            &cpu_buffer->new_pages, cpu_id)) {
1792                        err = -ENOMEM;
1793                        goto out_err;
1794                }
1795
1796                get_online_cpus();
1797
1798                /* Can't run something on an offline CPU. */
1799                if (!cpu_online(cpu_id))
1800                        rb_update_pages(cpu_buffer);
1801                else {
1802                        schedule_work_on(cpu_id,
1803                                         &cpu_buffer->update_pages_work);
1804                        wait_for_completion(&cpu_buffer->update_done);
1805                }
1806
1807                cpu_buffer->nr_pages_to_update = 0;
1808                put_online_cpus();
1809        }
1810
1811 out:
1812        /*
1813         * The ring buffer resize can happen with the ring buffer
1814         * enabled, so that the update disturbs the tracing as little
1815         * as possible. But if the buffer is disabled, we do not need
1816         * to worry about that, and we can take the time to verify
1817         * that the buffer is not corrupt.
1818         */
1819        if (atomic_read(&buffer->record_disabled)) {
1820                atomic_inc(&buffer->record_disabled);
1821                /*
1822                 * Even though the buffer was disabled, we must make sure
1823                 * that it is truly disabled before calling rb_check_pages.
1824                 * There could have been a race between checking
1825                 * record_disable and incrementing it.
1826                 */
1827                synchronize_sched();
1828                for_each_buffer_cpu(buffer, cpu) {
1829                        cpu_buffer = buffer->buffers[cpu];
1830                        rb_check_pages(cpu_buffer);
1831                }
1832                atomic_dec(&buffer->record_disabled);
1833        }
1834
1835        mutex_unlock(&buffer->mutex);
1836        return size;
1837
1838 out_err:
1839        for_each_buffer_cpu(buffer, cpu) {
1840                struct buffer_page *bpage, *tmp;
1841
1842                cpu_buffer = buffer->buffers[cpu];
1843                cpu_buffer->nr_pages_to_update = 0;
1844
1845                if (list_empty(&cpu_buffer->new_pages))
1846                        continue;
1847
1848                list_for_each_entry_safe(bpage, tmp, &cpu_buffer->new_pages,
1849                                        list) {
1850                        list_del_init(&bpage->list);
1851                        free_buffer_page(bpage);
1852                }
1853        }
1854        mutex_unlock(&buffer->mutex);
1855        return err;
1856}
1857EXPORT_SYMBOL_GPL(ring_buffer_resize);
1858
1859void ring_buffer_change_overwrite(struct ring_buffer *buffer, int val)
1860{
1861        mutex_lock(&buffer->mutex);
1862        if (val)
1863                buffer->flags |= RB_FL_OVERWRITE;
1864        else
1865                buffer->flags &= ~RB_FL_OVERWRITE;
1866        mutex_unlock(&buffer->mutex);
1867}
1868EXPORT_SYMBOL_GPL(ring_buffer_change_overwrite);
1869
1870static inline void *
1871__rb_data_page_index(struct buffer_data_page *bpage, unsigned index)
1872{
1873        return bpage->data + index;
1874}
1875
1876static inline void *__rb_page_index(struct buffer_page *bpage, unsigned index)
1877{
1878        return bpage->page->data + index;
1879}
1880
1881static inline struct ring_buffer_event *
1882rb_reader_event(struct ring_buffer_per_cpu *cpu_buffer)
1883{
1884        return __rb_page_index(cpu_buffer->reader_page,
1885                               cpu_buffer->reader_page->read);
1886}
1887
1888static inline struct ring_buffer_event *
1889rb_iter_head_event(struct ring_buffer_iter *iter)
1890{
1891        return __rb_page_index(iter->head_page, iter->head);
1892}
1893
1894static inline unsigned rb_page_commit(struct buffer_page *bpage)
1895{
1896        return local_read(&bpage->page->commit);
1897}
1898
1899/* Size is determined by what has been committed */
1900static inline unsigned rb_page_size(struct buffer_page *bpage)
1901{
1902        return rb_page_commit(bpage);
1903}
1904
1905static inline unsigned
1906rb_commit_index(struct ring_buffer_per_cpu *cpu_buffer)
1907{
1908        return rb_page_commit(cpu_buffer->commit_page);
1909}
1910
1911static inline unsigned
1912rb_event_index(struct ring_buffer_event *event)
1913{
1914        unsigned long addr = (unsigned long)event;
1915
1916        return (addr & ~PAGE_MASK) - BUF_PAGE_HDR_SIZE;
1917}
1918
1919static inline int
1920rb_event_is_commit(struct ring_buffer_per_cpu *cpu_buffer,
1921                   struct ring_buffer_event *event)
1922{
1923        unsigned long addr = (unsigned long)event;
1924        unsigned long index;
1925
1926        index = rb_event_index(event);
1927        addr &= PAGE_MASK;
1928
1929        return cpu_buffer->commit_page->page == (void *)addr &&
1930                rb_commit_index(cpu_buffer) == index;
1931}
1932
1933static void
1934rb_set_commit_to_write(struct ring_buffer_per_cpu *cpu_buffer)
1935{
1936        unsigned long max_count;
1937
1938        /*
1939         * We only race with interrupts and NMIs on this CPU.
1940         * If we own the commit event, then we can commit
1941         * all others that interrupted us, since the interruptions
1942         * are in stack format (they finish before they come
1943         * back to us). This allows us to do a simple loop to
1944         * assign the commit to the tail.
1945         */
1946 again:
1947        max_count = cpu_buffer->nr_pages * 100;
1948
1949        while (cpu_buffer->commit_page != cpu_buffer->tail_page) {
1950                if (RB_WARN_ON(cpu_buffer, !(--max_count)))
1951                        return;
1952                if (RB_WARN_ON(cpu_buffer,
1953                               rb_is_reader_page(cpu_buffer->tail_page)))
1954                        return;
1955                local_set(&cpu_buffer->commit_page->page->commit,
1956                          rb_page_write(cpu_buffer->commit_page));
1957                rb_inc_page(cpu_buffer, &cpu_buffer->commit_page);
1958                cpu_buffer->write_stamp =
1959                        cpu_buffer->commit_page->page->time_stamp;
1960                /* add barrier to keep gcc from optimizing too much */
1961                barrier();
1962        }
1963        while (rb_commit_index(cpu_buffer) !=
1964               rb_page_write(cpu_buffer->commit_page)) {
1965
1966                local_set(&cpu_buffer->commit_page->page->commit,
1967                          rb_page_write(cpu_buffer->commit_page));
1968                RB_WARN_ON(cpu_buffer,
1969                           local_read(&cpu_buffer->commit_page->page->commit) &
1970                           ~RB_WRITE_MASK);
1971                barrier();
1972        }
1973
1974        /* again, keep gcc from optimizing */
1975        barrier();
1976
1977        /*
1978         * If an interrupt came in just after the first while loop
1979         * and pushed the tail page forward, we will be left with
1980         * a dangling commit that will never go forward.
1981         */
1982        if (unlikely(cpu_buffer->commit_page != cpu_buffer->tail_page))
1983                goto again;
1984}
1985
1986static void rb_reset_reader_page(struct ring_buffer_per_cpu *cpu_buffer)
1987{
1988        cpu_buffer->read_stamp = cpu_buffer->reader_page->page->time_stamp;
1989        cpu_buffer->reader_page->read = 0;
1990}
1991
1992static void rb_inc_iter(struct ring_buffer_iter *iter)
1993{
1994        struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
1995
1996        /*
1997         * The iterator could be on the reader page (it starts there).
1998         * But the head could have moved, since the reader was
1999         * found. Check for this case and assign the iterator
2000         * to the head page instead of next.
2001         */
2002        if (iter->head_page == cpu_buffer->reader_page)
2003                iter->head_page = rb_set_head_page(cpu_buffer);
2004        else
2005                rb_inc_page(cpu_buffer, &iter->head_page);
2006
2007        iter->read_stamp = iter->head_page->page->time_stamp;
2008        iter->head = 0;
2009}
2010
2011/* Slow path, do not inline */
2012static noinline struct ring_buffer_event *
2013rb_add_time_stamp(struct ring_buffer_event *event, u64 delta)
2014{
2015        event->type_len = RINGBUF_TYPE_TIME_EXTEND;
2016
2017        /* Not the first event on the page? */
2018        if (rb_event_index(event)) {
2019                event->time_delta = delta & TS_MASK;
2020                event->array[0] = delta >> TS_SHIFT;
2021        } else {
2022                /* nope, just zero it */
2023                event->time_delta = 0;
2024                event->array[0] = 0;
2025        }
2026
2027        return skip_time_extend(event);
2028}
2029
2030/**
2031 * rb_update_event - update event type and data
2032 * @event: the event to update
2033 * @type: the type of event
2034 * @length: the size of the event field in the ring buffer
2035 *
2036 * Update the type and data fields of the event. The length
2037 * is the actual size that is written to the ring buffer,
2038 * and with this, we can determine what to place into the
2039 * data field.
2040 */
2041static void
2042rb_update_event(struct ring_buffer_per_cpu *cpu_buffer,
2043                struct ring_buffer_event *event, unsigned length,
2044                int add_timestamp, u64 delta)
2045{
2046        /* Only a commit updates the timestamp */
2047        if (unlikely(!rb_event_is_commit(cpu_buffer, event)))
2048                delta = 0;
2049
2050        /*
2051         * If we need to add a timestamp, then we
2052         * add it to the start of the resevered space.
2053         */
2054        if (unlikely(add_timestamp)) {
2055                event = rb_add_time_stamp(event, delta);
2056                length -= RB_LEN_TIME_EXTEND;
2057                delta = 0;
2058        }
2059
2060        event->time_delta = delta;
2061        length -= RB_EVNT_HDR_SIZE;
2062        if (length > RB_MAX_SMALL_DATA || RB_FORCE_8BYTE_ALIGNMENT) {
2063                event->type_len = 0;
2064                event->array[0] = length;
2065        } else
2066                event->type_len = DIV_ROUND_UP(length, RB_ALIGNMENT);
2067}
2068
2069/*
2070 * rb_handle_head_page - writer hit the head page
2071 *
2072 * Returns: +1 to retry page
2073 *           0 to continue
2074 *          -1 on error
2075 */
2076static int
2077rb_handle_head_page(struct ring_buffer_per_cpu *cpu_buffer,
2078                    struct buffer_page *tail_page,
2079                    struct buffer_page *next_page)
2080{
2081        struct buffer_page *new_head;
2082        int entries;
2083        int type;
2084        int ret;
2085
2086        entries = rb_page_entries(next_page);
2087
2088        /*
2089         * The hard part is here. We need to move the head
2090         * forward, and protect against both readers on
2091         * other CPUs and writers coming in via interrupts.
2092         */
2093        type = rb_head_page_set_update(cpu_buffer, next_page, tail_page,
2094                                       RB_PAGE_HEAD);
2095
2096        /*
2097         * type can be one of four:
2098         *  NORMAL - an interrupt already moved it for us
2099         *  HEAD   - we are the first to get here.
2100         *  UPDATE - we are the interrupt interrupting
2101         *           a current move.
2102         *  MOVED  - a reader on another CPU moved the next
2103         *           pointer to its reader page. Give up
2104         *           and try again.
2105         */
2106
2107        switch (type) {
2108        case RB_PAGE_HEAD:
2109                /*
2110                 * We changed the head to UPDATE, thus
2111                 * it is our responsibility to update
2112                 * the counters.
2113                 */
2114                local_add(entries, &cpu_buffer->overrun);
2115                local_sub(BUF_PAGE_SIZE, &cpu_buffer->entries_bytes);
2116
2117                /*
2118                 * The entries will be zeroed out when we move the
2119                 * tail page.
2120                 */
2121
2122                /* still more to do */
2123                break;
2124
2125        case RB_PAGE_UPDATE:
2126                /*
2127                 * This is an interrupt that interrupt the
2128                 * previous update. Still more to do.
2129                 */
2130                break;
2131        case RB_PAGE_NORMAL:
2132                /*
2133                 * An interrupt came in before the update
2134                 * and processed this for us.
2135                 * Nothing left to do.
2136                 */
2137                return 1;
2138        case RB_PAGE_MOVED:
2139                /*
2140                 * The reader is on another CPU and just did
2141                 * a swap with our next_page.
2142                 * Try again.
2143                 */
2144                return 1;
2145        default:
2146                RB_WARN_ON(cpu_buffer, 1); /* WTF??? */
2147                return -1;
2148        }
2149
2150        /*
2151         * Now that we are here, the old head pointer is
2152         * set to UPDATE. This will keep the reader from
2153         * swapping the head page with the reader page.
2154         * The reader (on another CPU) will spin till
2155         * we are finished.
2156         *
2157         * We just need to protect against interrupts
2158         * doing the job. We will set the next pointer
2159         * to HEAD. After that, we set the old pointer
2160         * to NORMAL, but only if it was HEAD before.
2161         * otherwise we are an interrupt, and only
2162         * want the outer most commit to reset it.
2163         */
2164        new_head = next_page;
2165        rb_inc_page(cpu_buffer, &new_head);
2166
2167        ret = rb_head_page_set_head(cpu_buffer, new_head, next_page,
2168                                    RB_PAGE_NORMAL);
2169
2170        /*
2171         * Valid returns are:
2172         *  HEAD   - an interrupt came in and already set it.
2173         *  NORMAL - One of two things:
2174         *            1) We really set it.
2175         *            2) A bunch of interrupts came in and moved
2176         *               the page forward again.
2177         */
2178        switch (ret) {
2179        case RB_PAGE_HEAD:
2180        case RB_PAGE_NORMAL:
2181                /* OK */
2182                break;
2183        default:
2184                RB_WARN_ON(cpu_buffer, 1);
2185                return -1;
2186        }
2187
2188        /*
2189         * It is possible that an interrupt came in,
2190         * set the head up, then more interrupts came in
2191         * and moved it again. When we get back here,
2192         * the page would have been set to NORMAL but we
2193         * just set it back to HEAD.
2194         *
2195         * How do you detect this? Well, if that happened
2196         * the tail page would have moved.
2197         */
2198        if (ret == RB_PAGE_NORMAL) {
2199                /*
2200                 * If the tail had moved passed next, then we need
2201                 * to reset the pointer.
2202                 */
2203                if (cpu_buffer->tail_page != tail_page &&
2204                    cpu_buffer->tail_page != next_page)
2205                        rb_head_page_set_normal(cpu_buffer, new_head,
2206                                                next_page,
2207                                                RB_PAGE_HEAD);
2208        }
2209
2210        /*
2211         * If this was the outer most commit (the one that
2212         * changed the original pointer from HEAD to UPDATE),
2213         * then it is up to us to reset it to NORMAL.
2214         */
2215        if (type == RB_PAGE_HEAD) {
2216                ret = rb_head_page_set_normal(cpu_buffer, next_page,
2217                                              tail_page,
2218                                              RB_PAGE_UPDATE);
2219                if (RB_WARN_ON(cpu_buffer,
2220                               ret != RB_PAGE_UPDATE))
2221                        return -1;
2222        }
2223
2224        return 0;
2225}
2226
2227static unsigned rb_calculate_event_length(unsigned length)
2228{
2229        struct ring_buffer_event event; /* Used only for sizeof array */
2230
2231        /* zero length can cause confusions */
2232        if (!length)
2233                length = 1;
2234
2235        if (length > RB_MAX_SMALL_DATA || RB_FORCE_8BYTE_ALIGNMENT)
2236                length += sizeof(event.array[0]);
2237
2238        length += RB_EVNT_HDR_SIZE;
2239        length = ALIGN(length, RB_ARCH_ALIGNMENT);
2240
2241        return length;
2242}
2243
2244static inline void
2245rb_reset_tail(struct ring_buffer_per_cpu *cpu_buffer,
2246              struct buffer_page *tail_page,
2247              unsigned long tail, unsigned long length)
2248{
2249        struct ring_buffer_event *event;
2250
2251        /*
2252         * Only the event that crossed the page boundary
2253         * must fill the old tail_page with padding.
2254         */
2255        if (tail >= BUF_PAGE_SIZE) {
2256                /*
2257                 * If the page was filled, then we still need
2258                 * to update the real_end. Reset it to zero
2259                 * and the reader will ignore it.
2260                 */
2261                if (tail == BUF_PAGE_SIZE)
2262                        tail_page->real_end = 0;
2263
2264                local_sub(length, &tail_page->write);
2265                return;
2266        }
2267
2268        event = __rb_page_index(tail_page, tail);
2269        kmemcheck_annotate_bitfield(event, bitfield);
2270
2271        /* account for padding bytes */
2272        local_add(BUF_PAGE_SIZE - tail, &cpu_buffer->entries_bytes);
2273
2274        /*
2275         * Save the original length to the meta data.
2276         * This will be used by the reader to add lost event
2277         * counter.
2278         */
2279        tail_page->real_end = tail;
2280
2281        /*
2282         * If this event is bigger than the minimum size, then
2283         * we need to be careful that we don't subtract the
2284         * write counter enough to allow another writer to slip
2285         * in on this page.
2286         * We put in a discarded commit instead, to make sure
2287         * that this space is not used again.
2288         *
2289         * If we are less than the minimum size, we don't need to
2290         * worry about it.
2291         */
2292        if (tail > (BUF_PAGE_SIZE - RB_EVNT_MIN_SIZE)) {
2293                /* No room for any events */
2294
2295                /* Mark the rest of the page with padding */
2296                rb_event_set_padding(event);
2297
2298                /* Set the write back to the previous setting */
2299                local_sub(length, &tail_page->write);
2300                return;
2301        }
2302
2303        /* Put in a discarded event */
2304        event->array[0] = (BUF_PAGE_SIZE - tail) - RB_EVNT_HDR_SIZE;
2305        event->type_len = RINGBUF_TYPE_PADDING;
2306        /* time delta must be non zero */
2307        event->time_delta = 1;
2308
2309        /* Set write to end of buffer */
2310        length = (tail + length) - BUF_PAGE_SIZE;
2311        local_sub(length, &tail_page->write);
2312}
2313
2314/*
2315 * This is the slow path, force gcc not to inline it.
2316 */
2317static noinline struct ring_buffer_event *
2318rb_move_tail(struct ring_buffer_per_cpu *cpu_buffer,
2319             unsigned long length, unsigned long tail,
2320             struct buffer_page *tail_page, u64 ts)
2321{
2322        struct buffer_page *commit_page = cpu_buffer->commit_page;
2323        struct ring_buffer *buffer = cpu_buffer->buffer;
2324        struct buffer_page *next_page;
2325        int ret;
2326
2327        next_page = tail_page;
2328
2329        rb_inc_page(cpu_buffer, &next_page);
2330
2331        /*
2332         * If for some reason, we had an interrupt storm that made
2333         * it all the way around the buffer, bail, and warn
2334         * about it.
2335         */
2336        if (unlikely(next_page == commit_page)) {
2337                local_inc(&cpu_buffer->commit_overrun);
2338                goto out_reset;
2339        }
2340
2341        /*
2342         * This is where the fun begins!
2343         *
2344         * We are fighting against races between a reader that
2345         * could be on another CPU trying to swap its reader
2346         * page with the buffer head.
2347         *
2348         * We are also fighting against interrupts coming in and
2349         * moving the head or tail on us as well.
2350         *
2351         * If the next page is the head page then we have filled
2352         * the buffer, unless the commit page is still on the
2353         * reader page.
2354         */
2355        if (rb_is_head_page(cpu_buffer, next_page, &tail_page->list)) {
2356
2357                /*
2358                 * If the commit is not on the reader page, then
2359                 * move the header page.
2360                 */
2361                if (!rb_is_reader_page(cpu_buffer->commit_page)) {
2362                        /*
2363                         * If we are not in overwrite mode,
2364                         * this is easy, just stop here.
2365                         */
2366                        if (!(buffer->flags & RB_FL_OVERWRITE)) {
2367                                local_inc(&cpu_buffer->dropped_events);
2368                                goto out_reset;
2369                        }
2370
2371                        ret = rb_handle_head_page(cpu_buffer,
2372                                                  tail_page,
2373                                                  next_page);
2374                        if (ret < 0)
2375                                goto out_reset;
2376                        if (ret)
2377                                goto out_again;
2378                } else {
2379                        /*
2380                         * We need to be careful here too. The
2381                         * commit page could still be on the reader
2382                         * page. We could have a small buffer, and
2383                         * have filled up the buffer with events
2384                         * from interrupts and such, and wrapped.
2385                         *
2386                         * Note, if the tail page is also the on the
2387                         * reader_page, we let it move out.
2388                         */
2389                        if (unlikely((cpu_buffer->commit_page !=
2390                                      cpu_buffer->tail_page) &&
2391                                     (cpu_buffer->commit_page ==
2392                                      cpu_buffer->reader_page))) {
2393                                local_inc(&cpu_buffer->commit_overrun);
2394                                goto out_reset;
2395                        }
2396                }
2397        }
2398
2399        ret = rb_tail_page_update(cpu_buffer, tail_page, next_page);
2400        if (ret) {
2401                /*
2402                 * Nested commits always have zero deltas, so
2403                 * just reread the time stamp
2404                 */
2405                ts = rb_time_stamp(buffer);
2406                next_page->page->time_stamp = ts;
2407        }
2408
2409 out_again:
2410
2411        rb_reset_tail(cpu_buffer, tail_page, tail, length);
2412
2413        /* fail and let the caller try again */
2414        return ERR_PTR(-EAGAIN);
2415
2416 out_reset:
2417        /* reset write */
2418        rb_reset_tail(cpu_buffer, tail_page, tail, length);
2419
2420        return NULL;
2421}
2422
2423static struct ring_buffer_event *
2424__rb_reserve_next(struct ring_buffer_per_cpu *cpu_buffer,
2425                  unsigned long length, u64 ts,
2426                  u64 delta, int add_timestamp)
2427{
2428        struct buffer_page *tail_page;
2429        struct ring_buffer_event *event;
2430        unsigned long tail, write;
2431
2432        /*
2433         * If the time delta since the last event is too big to
2434         * hold in the time field of the event, then we append a
2435         * TIME EXTEND event ahead of the data event.
2436         */
2437        if (unlikely(add_timestamp))
2438                length += RB_LEN_TIME_EXTEND;
2439
2440        tail_page = cpu_buffer->tail_page;
2441        write = local_add_return(length, &tail_page->write);
2442
2443        /* set write to only the index of the write */
2444        write &= RB_WRITE_MASK;
2445        tail = write - length;
2446
2447        /*
2448         * If this is the first commit on the page, then it has the same
2449         * timestamp as the page itself.
2450         */
2451        if (!tail)
2452                delta = 0;
2453
2454        /* See if we shot pass the end of this buffer page */
2455        if (unlikely(write > BUF_PAGE_SIZE))
2456                return rb_move_tail(cpu_buffer, length, tail,
2457                                    tail_page, ts);
2458
2459        /* We reserved something on the buffer */
2460
2461        event = __rb_page_index(tail_page, tail);
2462        kmemcheck_annotate_bitfield(event, bitfield);
2463        rb_update_event(cpu_buffer, event, length, add_timestamp, delta);
2464
2465        local_inc(&tail_page->entries);
2466
2467        /*
2468         * If this is the first commit on the page, then update
2469         * its timestamp.
2470         */
2471        if (!tail)
2472                tail_page->page->time_stamp = ts;
2473
2474        /* account for these added bytes */
2475        local_add(length, &cpu_buffer->entries_bytes);
2476
2477        return event;
2478}
2479
2480static inline int
2481rb_try_to_discard(struct ring_buffer_per_cpu *cpu_buffer,
2482                  struct ring_buffer_event *event)
2483{
2484        unsigned long new_index, old_index;
2485        struct buffer_page *bpage;
2486        unsigned long index;
2487        unsigned long addr;
2488
2489        new_index = rb_event_index(event);
2490        old_index = new_index + rb_event_ts_length(event);
2491        addr = (unsigned long)event;
2492        addr &= PAGE_MASK;
2493
2494        bpage = cpu_buffer->tail_page;
2495
2496        if (bpage->page == (void *)addr && rb_page_write(bpage) == old_index) {
2497                unsigned long write_mask =
2498                        local_read(&bpage->write) & ~RB_WRITE_MASK;
2499                unsigned long event_length = rb_event_length(event);
2500                /*
2501                 * This is on the tail page. It is possible that
2502                 * a write could come in and move the tail page
2503                 * and write to the next page. That is fine
2504                 * because we just shorten what is on this page.
2505                 */
2506                old_index += write_mask;
2507                new_index += write_mask;
2508                index = local_cmpxchg(&bpage->write, old_index, new_index);
2509                if (index == old_index) {
2510                        /* update counters */
2511                        local_sub(event_length, &cpu_buffer->entries_bytes);
2512                        return 1;
2513                }
2514        }
2515
2516        /* could not discard */
2517        return 0;
2518}
2519
2520static void rb_start_commit(struct ring_buffer_per_cpu *cpu_buffer)
2521{
2522        local_inc(&cpu_buffer->committing);
2523        local_inc(&cpu_buffer->commits);
2524}
2525
2526static inline void rb_end_commit(struct ring_buffer_per_cpu *cpu_buffer)
2527{
2528        unsigned long commits;
2529
2530        if (RB_WARN_ON(cpu_buffer,
2531                       !local_read(&cpu_buffer->committing)))
2532                return;
2533
2534 again:
2535        commits = local_read(&cpu_buffer->commits);
2536        /* synchronize with interrupts */
2537        barrier();
2538        if (local_read(&cpu_buffer->committing) == 1)
2539                rb_set_commit_to_write(cpu_buffer);
2540
2541        local_dec(&cpu_buffer->committing);
2542
2543        /* synchronize with interrupts */
2544        barrier();
2545
2546        /*
2547         * Need to account for interrupts coming in between the
2548         * updating of the commit page and the clearing of the
2549         * committing counter.
2550         */
2551        if (unlikely(local_read(&cpu_buffer->commits) != commits) &&
2552            !local_read(&cpu_buffer->committing)) {
2553                local_inc(&cpu_buffer->committing);
2554                goto again;
2555        }
2556}
2557
2558static struct ring_buffer_event *
2559rb_reserve_next_event(struct ring_buffer *buffer,
2560                      struct ring_buffer_per_cpu *cpu_buffer,
2561                      unsigned long length)
2562{
2563        struct ring_buffer_event *event;
2564        u64 ts, delta;
2565        int nr_loops = 0;
2566        int add_timestamp;
2567        u64 diff;
2568
2569        rb_start_commit(cpu_buffer);
2570
2571#ifdef CONFIG_RING_BUFFER_ALLOW_SWAP
2572        /*
2573         * Due to the ability to swap a cpu buffer from a buffer
2574         * it is possible it was swapped before we committed.
2575         * (committing stops a swap). We check for it here and
2576         * if it happened, we have to fail the write.
2577         */
2578        barrier();
2579        if (unlikely(ACCESS_ONCE(cpu_buffer->buffer) != buffer)) {
2580                local_dec(&cpu_buffer->committing);
2581                local_dec(&cpu_buffer->commits);
2582                return NULL;
2583        }
2584#endif
2585
2586        length = rb_calculate_event_length(length);
2587 again:
2588        add_timestamp = 0;
2589        delta = 0;
2590
2591        /*
2592         * We allow for interrupts to reenter here and do a trace.
2593         * If one does, it will cause this original code to loop
2594         * back here. Even with heavy interrupts happening, this
2595         * should only happen a few times in a row. If this happens
2596         * 1000 times in a row, there must be either an interrupt
2597         * storm or we have something buggy.
2598         * Bail!
2599         */
2600        if (RB_WARN_ON(cpu_buffer, ++nr_loops > 1000))
2601                goto out_fail;
2602
2603        ts = rb_time_stamp(cpu_buffer->buffer);
2604        diff = ts - cpu_buffer->write_stamp;
2605
2606        /* make sure this diff is calculated here */
2607        barrier();
2608
2609        /* Did the write stamp get updated already? */
2610        if (likely(ts >= cpu_buffer->write_stamp)) {
2611                delta = diff;
2612                if (unlikely(test_time_stamp(delta))) {
2613                        int local_clock_stable = 1;
2614#ifdef CONFIG_HAVE_UNSTABLE_SCHED_CLOCK
2615                        local_clock_stable = sched_clock_stable();
2616#endif
2617                        WARN_ONCE(delta > (1ULL << 59),
2618                                  KERN_WARNING "Delta way too big! %llu ts=%llu write stamp = %llu\n%s",
2619                                  (unsigned long long)delta,
2620                                  (unsigned long long)ts,
2621                                  (unsigned long long)cpu_buffer->write_stamp,
2622                                  local_clock_stable ? "" :
2623                                  "If you just came from a suspend/resume,\n"
2624                                  "please switch to the trace global clock:\n"
2625                                  "  echo global > /sys/kernel/debug/tracing/trace_clock\n");
2626                        add_timestamp = 1;
2627                }
2628        }
2629
2630        event = __rb_reserve_next(cpu_buffer, length, ts,
2631                                  delta, add_timestamp);
2632        if (unlikely(PTR_ERR(event) == -EAGAIN))
2633                goto again;
2634
2635        if (!event)
2636                goto out_fail;
2637
2638        return event;
2639
2640 out_fail:
2641        rb_end_commit(cpu_buffer);
2642        return NULL;
2643}
2644
2645#ifdef CONFIG_TRACING
2646
2647/*
2648 * The lock and unlock are done within a preempt disable section.
2649 * The current_context per_cpu variable can only be modified
2650 * by the current task between lock and unlock. But it can
2651 * be modified more than once via an interrupt. To pass this
2652 * information from the lock to the unlock without having to
2653 * access the 'in_interrupt()' functions again (which do show
2654 * a bit of overhead in something as critical as function tracing,
2655 * we use a bitmask trick.
2656 *
2657 *  bit 0 =  NMI context
2658 *  bit 1 =  IRQ context
2659 *  bit 2 =  SoftIRQ context
2660 *  bit 3 =  normal context.
2661 *
2662 * This works because this is the order of contexts that can
2663 * preempt other contexts. A SoftIRQ never preempts an IRQ
2664 * context.
2665 *
2666 * When the context is determined, the corresponding bit is
2667 * checked and set (if it was set, then a recursion of that context
2668 * happened).
2669 *
2670 * On unlock, we need to clear this bit. To do so, just subtract
2671 * 1 from the current_context and AND it to itself.
2672 *
2673 * (binary)
2674 *  101 - 1 = 100
2675 *  101 & 100 = 100 (clearing bit zero)
2676 *
2677 *  1010 - 1 = 1001
2678 *  1010 & 1001 = 1000 (clearing bit 1)
2679 *
2680 * The least significant bit can be cleared this way, and it
2681 * just so happens that it is the same bit corresponding to
2682 * the current context.
2683 */
2684
2685static __always_inline int
2686trace_recursive_lock(struct ring_buffer_per_cpu *cpu_buffer)
2687{
2688        unsigned int val = cpu_buffer->current_context;
2689        int bit;
2690
2691        if (in_interrupt()) {
2692                if (in_nmi())
2693                        bit = 0;
2694                else if (in_irq())
2695                        bit = 1;
2696                else
2697                        bit = 2;
2698        } else
2699                bit = 3;
2700
2701        if (unlikely(val & (1 << bit)))
2702                return 1;
2703
2704        val |= (1 << bit);
2705        cpu_buffer->current_context = val;
2706
2707        return 0;
2708}
2709
2710static __always_inline void
2711trace_recursive_unlock(struct ring_buffer_per_cpu *cpu_buffer)
2712{
2713        cpu_buffer->current_context &= cpu_buffer->current_context - 1;
2714}
2715
2716#else
2717
2718#define trace_recursive_lock(cpu_buffer)        (0)
2719#define trace_recursive_unlock(cpu_buffer)      do { } while (0)
2720
2721#endif
2722
2723/**
2724 * ring_buffer_lock_reserve - reserve a part of the buffer
2725 * @buffer: the ring buffer to reserve from
2726 * @length: the length of the data to reserve (excluding event header)
2727 *
2728 * Returns a reseverd event on the ring buffer to copy directly to.
2729 * The user of this interface will need to get the body to write into
2730 * and can use the ring_buffer_event_data() interface.
2731 *
2732 * The length is the length of the data needed, not the event length
2733 * which also includes the event header.
2734 *
2735 * Must be paired with ring_buffer_unlock_commit, unless NULL is returned.
2736 * If NULL is returned, then nothing has been allocated or locked.
2737 */
2738struct ring_buffer_event *
2739ring_buffer_lock_reserve(struct ring_buffer *buffer, unsigned long length)
2740{
2741        struct ring_buffer_per_cpu *cpu_buffer;
2742        struct ring_buffer_event *event;
2743        int cpu;
2744
2745        if (ring_buffer_flags != RB_BUFFERS_ON)
2746                return NULL;
2747
2748        /* If we are tracing schedule, we don't want to recurse */
2749        preempt_disable_notrace();
2750
2751        if (unlikely(atomic_read(&buffer->record_disabled)))
2752                goto out;
2753
2754        cpu = raw_smp_processor_id();
2755
2756        if (unlikely(!cpumask_test_cpu(cpu, buffer->cpumask)))
2757                goto out;
2758
2759        cpu_buffer = buffer->buffers[cpu];
2760
2761        if (unlikely(atomic_read(&cpu_buffer->record_disabled)))
2762                goto out;
2763
2764        if (unlikely(length > BUF_MAX_DATA_SIZE))
2765                goto out;
2766
2767        if (unlikely(trace_recursive_lock(cpu_buffer)))
2768                goto out;
2769
2770        event = rb_reserve_next_event(buffer, cpu_buffer, length);
2771        if (!event)
2772                goto out_unlock;
2773
2774        return event;
2775
2776 out_unlock:
2777        trace_recursive_unlock(cpu_buffer);
2778 out:
2779        preempt_enable_notrace();
2780        return NULL;
2781}
2782EXPORT_SYMBOL_GPL(ring_buffer_lock_reserve);
2783
2784static void
2785rb_update_write_stamp(struct ring_buffer_per_cpu *cpu_buffer,
2786                      struct ring_buffer_event *event)
2787{
2788        u64 delta;
2789
2790        /*
2791         * The event first in the commit queue updates the
2792         * time stamp.
2793         */
2794        if (rb_event_is_commit(cpu_buffer, event)) {
2795                /*
2796                 * A commit event that is first on a page
2797                 * updates the write timestamp with the page stamp
2798                 */
2799                if (!rb_event_index(event))
2800                        cpu_buffer->write_stamp =
2801                                cpu_buffer->commit_page->page->time_stamp;
2802                else if (event->type_len == RINGBUF_TYPE_TIME_EXTEND) {
2803                        delta = event->array[0];
2804                        delta <<= TS_SHIFT;
2805                        delta += event->time_delta;
2806                        cpu_buffer->write_stamp += delta;
2807                } else
2808                        cpu_buffer->write_stamp += event->time_delta;
2809        }
2810}
2811
2812static void rb_commit(struct ring_buffer_per_cpu *cpu_buffer,
2813                      struct ring_buffer_event *event)
2814{
2815        local_inc(&cpu_buffer->entries);
2816        rb_update_write_stamp(cpu_buffer, event);
2817        rb_end_commit(cpu_buffer);
2818}
2819
2820static __always_inline void
2821rb_wakeups(struct ring_buffer *buffer, struct ring_buffer_per_cpu *cpu_buffer)
2822{
2823        bool pagebusy;
2824
2825        if (buffer->irq_work.waiters_pending) {
2826                buffer->irq_work.waiters_pending = false;
2827                /* irq_work_queue() supplies it's own memory barriers */
2828                irq_work_queue(&buffer->irq_work.work);
2829        }
2830
2831        if (cpu_buffer->irq_work.waiters_pending) {
2832                cpu_buffer->irq_work.waiters_pending = false;
2833                /* irq_work_queue() supplies it's own memory barriers */
2834                irq_work_queue(&cpu_buffer->irq_work.work);
2835        }
2836
2837        pagebusy = cpu_buffer->reader_page == cpu_buffer->commit_page;
2838
2839        if (!pagebusy && cpu_buffer->irq_work.full_waiters_pending) {
2840                cpu_buffer->irq_work.wakeup_full = true;
2841                cpu_buffer->irq_work.full_waiters_pending = false;
2842                /* irq_work_queue() supplies it's own memory barriers */
2843                irq_work_queue(&cpu_buffer->irq_work.work);
2844        }
2845}
2846
2847/**
2848 * ring_buffer_unlock_commit - commit a reserved
2849 * @buffer: The buffer to commit to
2850 * @event: The event pointer to commit.
2851 *
2852 * This commits the data to the ring buffer, and releases any locks held.
2853 *
2854 * Must be paired with ring_buffer_lock_reserve.
2855 */
2856int ring_buffer_unlock_commit(struct ring_buffer *buffer,
2857                              struct ring_buffer_event *event)
2858{
2859        struct ring_buffer_per_cpu *cpu_buffer;
2860        int cpu = raw_smp_processor_id();
2861
2862        cpu_buffer = buffer->buffers[cpu];
2863
2864        rb_commit(cpu_buffer, event);
2865
2866        rb_wakeups(buffer, cpu_buffer);
2867
2868        trace_recursive_unlock(cpu_buffer);
2869
2870        preempt_enable_notrace();
2871
2872        return 0;
2873}
2874EXPORT_SYMBOL_GPL(ring_buffer_unlock_commit);
2875
2876static inline void rb_event_discard(struct ring_buffer_event *event)
2877{
2878        if (event->type_len == RINGBUF_TYPE_TIME_EXTEND)
2879                event = skip_time_extend(event);
2880
2881        /* array[0] holds the actual length for the discarded event */
2882        event->array[0] = rb_event_data_length(event) - RB_EVNT_HDR_SIZE;
2883        event->type_len = RINGBUF_TYPE_PADDING;
2884        /* time delta must be non zero */
2885        if (!event->time_delta)
2886                event->time_delta = 1;
2887}
2888
2889/*
2890 * Decrement the entries to the page that an event is on.
2891 * The event does not even need to exist, only the pointer
2892 * to the page it is on. This may only be called before the commit
2893 * takes place.
2894 */
2895static inline void
2896rb_decrement_entry(struct ring_buffer_per_cpu *cpu_buffer,
2897                   struct ring_buffer_event *event)
2898{
2899        unsigned long addr = (unsigned long)event;
2900        struct buffer_page *bpage = cpu_buffer->commit_page;
2901        struct buffer_page *start;
2902
2903        addr &= PAGE_MASK;
2904
2905        /* Do the likely case first */
2906        if (likely(bpage->page == (void *)addr)) {
2907                local_dec(&bpage->entries);
2908                return;
2909        }
2910
2911        /*
2912         * Because the commit page may be on the reader page we
2913         * start with the next page and check the end loop there.
2914         */
2915        rb_inc_page(cpu_buffer, &bpage);
2916        start = bpage;
2917        do {
2918                if (bpage->page == (void *)addr) {
2919                        local_dec(&bpage->entries);
2920                        return;
2921                }
2922                rb_inc_page(cpu_buffer, &bpage);
2923        } while (bpage != start);
2924
2925        /* commit not part of this buffer?? */
2926        RB_WARN_ON(cpu_buffer, 1);
2927}
2928
2929/**
2930 * ring_buffer_commit_discard - discard an event that has not been committed
2931 * @buffer: the ring buffer
2932 * @event: non committed event to discard
2933 *
2934 * Sometimes an event that is in the ring buffer needs to be ignored.
2935 * This function lets the user discard an event in the ring buffer
2936 * and then that event will not be read later.
2937 *
2938 * This function only works if it is called before the the item has been
2939 * committed. It will try to free the event from the ring buffer
2940 * if another event has not been added behind it.
2941 *
2942 * If another event has been added behind it, it will set the event
2943 * up as discarded, and perform the commit.
2944 *
2945 * If this function is called, do not call ring_buffer_unlock_commit on
2946 * the event.
2947 */
2948void ring_buffer_discard_commit(struct ring_buffer *buffer,
2949                                struct ring_buffer_event *event)
2950{
2951        struct ring_buffer_per_cpu *cpu_buffer;
2952        int cpu;
2953
2954        /* The event is discarded regardless */
2955        rb_event_discard(event);
2956
2957        cpu = smp_processor_id();
2958        cpu_buffer = buffer->buffers[cpu];
2959
2960        /*
2961         * This must only be called if the event has not been
2962         * committed yet. Thus we can assume that preemption
2963         * is still disabled.
2964         */
2965        RB_WARN_ON(buffer, !local_read(&cpu_buffer->committing));
2966
2967        rb_decrement_entry(cpu_buffer, event);
2968        if (rb_try_to_discard(cpu_buffer, event))
2969                goto out;
2970
2971        /*
2972         * The commit is still visible by the reader, so we
2973         * must still update the timestamp.
2974         */
2975        rb_update_write_stamp(cpu_buffer, event);
2976 out:
2977        rb_end_commit(cpu_buffer);
2978
2979        trace_recursive_unlock(cpu_buffer);
2980
2981        preempt_enable_notrace();
2982
2983}
2984EXPORT_SYMBOL_GPL(ring_buffer_discard_commit);
2985
2986/**
2987 * ring_buffer_write - write data to the buffer without reserving
2988 * @buffer: The ring buffer to write to.
2989 * @length: The length of the data being written (excluding the event header)
2990 * @data: The data to write to the buffer.
2991 *
2992 * This is like ring_buffer_lock_reserve and ring_buffer_unlock_commit as
2993 * one function. If you already have the data to write to the buffer, it
2994 * may be easier to simply call this function.
2995 *
2996 * Note, like ring_buffer_lock_reserve, the length is the length of the data
2997 * and not the length of the event which would hold the header.
2998 */
2999int ring_buffer_write(struct ring_buffer *buffer,
3000                      unsigned long length,
3001                      void *data)
3002{
3003        struct ring_buffer_per_cpu *cpu_buffer;
3004        struct ring_buffer_event *event;
3005        void *body;
3006        int ret = -EBUSY;
3007        int cpu;
3008
3009        if (ring_buffer_flags != RB_BUFFERS_ON)
3010                return -EBUSY;
3011
3012        preempt_disable_notrace();
3013
3014        if (atomic_read(&buffer->record_disabled))
3015                goto out;
3016
3017        cpu = raw_smp_processor_id();
3018
3019        if (!cpumask_test_cpu(cpu, buffer->cpumask))
3020                goto out;
3021
3022        cpu_buffer = buffer->buffers[cpu];
3023
3024        if (atomic_read(&cpu_buffer->record_disabled))
3025                goto out;
3026
3027        if (length > BUF_MAX_DATA_SIZE)
3028                goto out;
3029
3030        event = rb_reserve_next_event(buffer, cpu_buffer, length);
3031        if (!event)
3032                goto out;
3033
3034        body = rb_event_data(event);
3035
3036        memcpy(body, data, length);
3037
3038        rb_commit(cpu_buffer, event);
3039
3040        rb_wakeups(buffer, cpu_buffer);
3041
3042        ret = 0;
3043 out:
3044        preempt_enable_notrace();
3045
3046        return ret;
3047}
3048EXPORT_SYMBOL_GPL(ring_buffer_write);
3049
3050static int rb_per_cpu_empty(struct ring_buffer_per_cpu *cpu_buffer)
3051{
3052        struct buffer_page *reader = cpu_buffer->reader_page;
3053        struct buffer_page *head = rb_set_head_page(cpu_buffer);
3054        struct buffer_page *commit = cpu_buffer->commit_page;
3055
3056        /* In case of error, head will be NULL */
3057        if (unlikely(!head))
3058                return 1;
3059
3060        return reader->read == rb_page_commit(reader) &&
3061                (commit == reader ||
3062                 (commit == head &&
3063                  head->read == rb_page_commit(commit)));
3064}
3065
3066/**
3067 * ring_buffer_record_disable - stop all writes into the buffer
3068 * @buffer: The ring buffer to stop writes to.
3069 *
3070 * This prevents all writes to the buffer. Any attempt to write
3071 * to the buffer after this will fail and return NULL.
3072 *
3073 * The caller should call synchronize_sched() after this.
3074 */
3075void ring_buffer_record_disable(struct ring_buffer *buffer)
3076{
3077        atomic_inc(&buffer->record_disabled);
3078}
3079EXPORT_SYMBOL_GPL(ring_buffer_record_disable);
3080
3081/**
3082 * ring_buffer_record_enable - enable writes to the buffer
3083 * @buffer: The ring buffer to enable writes
3084 *
3085 * Note, multiple disables will need the same number of enables
3086 * to truly enable the writing (much like preempt_disable).
3087 */
3088void ring_buffer_record_enable(struct ring_buffer *buffer)
3089{
3090        atomic_dec(&buffer->record_disabled);
3091}
3092EXPORT_SYMBOL_GPL(ring_buffer_record_enable);
3093
3094/**
3095 * ring_buffer_record_off - stop all writes into the buffer
3096 * @buffer: The ring buffer to stop writes to.
3097 *
3098 * This prevents all writes to the buffer. Any attempt to write
3099 * to the buffer after this will fail and return NULL.
3100 *
3101 * This is different than ring_buffer_record_disable() as
3102 * it works like an on/off switch, where as the disable() version
3103 * must be paired with a enable().
3104 */
3105void ring_buffer_record_off(struct ring_buffer *buffer)
3106{
3107        unsigned int rd;
3108        unsigned int new_rd;
3109
3110        do {
3111                rd = atomic_read(&buffer->record_disabled);
3112                new_rd = rd | RB_BUFFER_OFF;
3113        } while (atomic_cmpxchg(&buffer->record_disabled, rd, new_rd) != rd);
3114}
3115EXPORT_SYMBOL_GPL(ring_buffer_record_off);
3116
3117/**
3118 * ring_buffer_record_on - restart writes into the buffer
3119 * @buffer: The ring buffer to start writes to.
3120 *
3121 * This enables all writes to the buffer that was disabled by
3122 * ring_buffer_record_off().
3123 *
3124 * This is different than ring_buffer_record_enable() as
3125 * it works like an on/off switch, where as the enable() version
3126 * must be paired with a disable().
3127 */
3128void ring_buffer_record_on(struct ring_buffer *buffer)
3129{
3130        unsigned int rd;
3131        unsigned int new_rd;
3132
3133        do {
3134                rd = atomic_read(&buffer->record_disabled);
3135                new_rd = rd & ~RB_BUFFER_OFF;
3136        } while (atomic_cmpxchg(&buffer->record_disabled, rd, new_rd) != rd);
3137}
3138EXPORT_SYMBOL_GPL(ring_buffer_record_on);
3139
3140/**
3141 * ring_buffer_record_is_on - return true if the ring buffer can write
3142 * @buffer: The ring buffer to see if write is enabled
3143 *
3144 * Returns true if the ring buffer is in a state that it accepts writes.
3145 */
3146int ring_buffer_record_is_on(struct ring_buffer *buffer)
3147{
3148        return !atomic_read(&buffer->record_disabled);
3149}
3150
3151/**
3152 * ring_buffer_record_disable_cpu - stop all writes into the cpu_buffer
3153 * @buffer: The ring buffer to stop writes to.
3154 * @cpu: The CPU buffer to stop
3155 *
3156 * This prevents all writes to the buffer. Any attempt to write
3157 * to the buffer after this will fail and return NULL.
3158 *
3159 * The caller should call synchronize_sched() after this.
3160 */
3161void ring_buffer_record_disable_cpu(struct ring_buffer *buffer, int cpu)
3162{
3163        struct ring_buffer_per_cpu *cpu_buffer;
3164
3165        if (!cpumask_test_cpu(cpu, buffer->cpumask))
3166                return;
3167
3168        cpu_buffer = buffer->buffers[cpu];
3169        atomic_inc(&cpu_buffer->record_disabled);
3170}
3171EXPORT_SYMBOL_GPL(ring_buffer_record_disable_cpu);
3172
3173/**
3174 * ring_buffer_record_enable_cpu - enable writes to the buffer
3175 * @buffer: The ring buffer to enable writes
3176 * @cpu: The CPU to enable.
3177 *
3178 * Note, multiple disables will need the same number of enables
3179 * to truly enable the writing (much like preempt_disable).
3180 */
3181void ring_buffer_record_enable_cpu(struct ring_buffer *buffer, int cpu)
3182{
3183        struct ring_buffer_per_cpu *cpu_buffer;
3184
3185        if (!cpumask_test_cpu(cpu, buffer->cpumask))
3186                return;
3187
3188        cpu_buffer = buffer->buffers[cpu];
3189        atomic_dec(&cpu_buffer->record_disabled);
3190}
3191EXPORT_SYMBOL_GPL(ring_buffer_record_enable_cpu);
3192
3193/*
3194 * The total entries in the ring buffer is the running counter
3195 * of entries entered into the ring buffer, minus the sum of
3196 * the entries read from the ring buffer and the number of
3197 * entries that were overwritten.
3198 */
3199static inline unsigned long
3200rb_num_of_entries(struct ring_buffer_per_cpu *cpu_buffer)
3201{
3202        return local_read(&cpu_buffer->entries) -
3203                (local_read(&cpu_buffer->overrun) + cpu_buffer->read);
3204}
3205
3206/**
3207 * ring_buffer_oldest_event_ts - get the oldest event timestamp from the buffer
3208 * @buffer: The ring buffer
3209 * @cpu: The per CPU buffer to read from.
3210 */
3211u64 ring_buffer_oldest_event_ts(struct ring_buffer *buffer, int cpu)
3212{
3213        unsigned long flags;
3214        struct ring_buffer_per_cpu *cpu_buffer;
3215        struct buffer_page *bpage;
3216        u64 ret = 0;
3217
3218        if (!cpumask_test_cpu(cpu, buffer->cpumask))
3219                return 0;
3220
3221        cpu_buffer = buffer->buffers[cpu];
3222        raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
3223        /*
3224         * if the tail is on reader_page, oldest time stamp is on the reader
3225         * page
3226         */
3227        if (cpu_buffer->tail_page == cpu_buffer->reader_page)
3228                bpage = cpu_buffer->reader_page;
3229        else
3230                bpage = rb_set_head_page(cpu_buffer);
3231        if (bpage)
3232                ret = bpage->page->time_stamp;
3233        raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
3234
3235        return ret;
3236}
3237EXPORT_SYMBOL_GPL(ring_buffer_oldest_event_ts);
3238
3239/**
3240 * ring_buffer_bytes_cpu - get the number of bytes consumed in a cpu buffer
3241 * @buffer: The ring buffer
3242 * @cpu: The per CPU buffer to read from.
3243 */
3244unsigned long ring_buffer_bytes_cpu(struct ring_buffer *buffer, int cpu)
3245{
3246        struct ring_buffer_per_cpu *cpu_buffer;
3247        unsigned long ret;
3248
3249        if (!cpumask_test_cpu(cpu, buffer->cpumask))
3250                return 0;
3251
3252        cpu_buffer = buffer->buffers[cpu];
3253        ret = local_read(&cpu_buffer->entries_bytes) - cpu_buffer->read_bytes;
3254
3255        return ret;
3256}
3257EXPORT_SYMBOL_GPL(ring_buffer_bytes_cpu);
3258
3259/**
3260 * ring_buffer_entries_cpu - get the number of entries in a cpu buffer
3261 * @buffer: The ring buffer
3262 * @cpu: The per CPU buffer to get the entries from.
3263 */
3264unsigned long ring_buffer_entries_cpu(struct ring_buffer *buffer, int cpu)
3265{
3266        struct ring_buffer_per_cpu *cpu_buffer;
3267
3268        if (!cpumask_test_cpu(cpu, buffer->cpumask))
3269                return 0;
3270
3271        cpu_buffer = buffer->buffers[cpu];
3272
3273        return rb_num_of_entries(cpu_buffer);
3274}
3275EXPORT_SYMBOL_GPL(ring_buffer_entries_cpu);
3276
3277/**
3278 * ring_buffer_overrun_cpu - get the number of overruns caused by the ring
3279 * buffer wrapping around (only if RB_FL_OVERWRITE is on).
3280 * @buffer: The ring buffer
3281 * @cpu: The per CPU buffer to get the number of overruns from
3282 */
3283unsigned long ring_buffer_overrun_cpu(struct ring_buffer *buffer, int cpu)
3284{
3285        struct ring_buffer_per_cpu *cpu_buffer;
3286        unsigned long ret;
3287
3288        if (!cpumask_test_cpu(cpu, buffer->cpumask))
3289                return 0;
3290
3291        cpu_buffer = buffer->buffers[cpu];
3292        ret = local_read(&cpu_buffer->overrun);
3293
3294        return ret;
3295}
3296EXPORT_SYMBOL_GPL(ring_buffer_overrun_cpu);
3297
3298/**
3299 * ring_buffer_commit_overrun_cpu - get the number of overruns caused by
3300 * commits failing due to the buffer wrapping around while there are uncommitted
3301 * events, such as during an interrupt storm.
3302 * @buffer: The ring buffer
3303 * @cpu: The per CPU buffer to get the number of overruns from
3304 */
3305unsigned long
3306ring_buffer_commit_overrun_cpu(struct ring_buffer *buffer, int cpu)
3307{
3308        struct ring_buffer_per_cpu *cpu_buffer;
3309        unsigned long ret;
3310
3311        if (!cpumask_test_cpu(cpu, buffer->cpumask))
3312                return 0;
3313
3314        cpu_buffer = buffer->buffers[cpu];
3315        ret = local_read(&cpu_buffer->commit_overrun);
3316
3317        return ret;
3318}
3319EXPORT_SYMBOL_GPL(ring_buffer_commit_overrun_cpu);
3320
3321/**
3322 * ring_buffer_dropped_events_cpu - get the number of dropped events caused by
3323 * the ring buffer filling up (only if RB_FL_OVERWRITE is off).
3324 * @buffer: The ring buffer
3325 * @cpu: The per CPU buffer to get the number of overruns from
3326 */
3327unsigned long
3328ring_buffer_dropped_events_cpu(struct ring_buffer *buffer, int cpu)
3329{
3330        struct ring_buffer_per_cpu *cpu_buffer;
3331        unsigned long ret;
3332
3333        if (!cpumask_test_cpu(cpu, buffer->cpumask))
3334                return 0;
3335
3336        cpu_buffer = buffer->buffers[cpu];
3337        ret = local_read(&cpu_buffer->dropped_events);
3338
3339        return ret;
3340}
3341EXPORT_SYMBOL_GPL(ring_buffer_dropped_events_cpu);
3342
3343/**
3344 * ring_buffer_read_events_cpu - get the number of events successfully read
3345 * @buffer: The ring buffer
3346 * @cpu: The per CPU buffer to get the number of events read
3347 */
3348unsigned long
3349ring_buffer_read_events_cpu(struct ring_buffer *buffer, int cpu)
3350{
3351        struct ring_buffer_per_cpu *cpu_buffer;
3352
3353        if (!cpumask_test_cpu(cpu, buffer->cpumask))
3354                return 0;
3355
3356        cpu_buffer = buffer->buffers[cpu];
3357        return cpu_buffer->read;
3358}
3359EXPORT_SYMBOL_GPL(ring_buffer_read_events_cpu);
3360
3361/**
3362 * ring_buffer_entries - get the number of entries in a buffer
3363 * @buffer: The ring buffer
3364 *
3365 * Returns the total number of entries in the ring buffer
3366 * (all CPU entries)
3367 */
3368unsigned long ring_buffer_entries(struct ring_buffer *buffer)
3369{
3370        struct ring_buffer_per_cpu *cpu_buffer;
3371        unsigned long entries = 0;
3372        int cpu;
3373
3374        /* if you care about this being correct, lock the buffer */
3375        for_each_buffer_cpu(buffer, cpu) {
3376                cpu_buffer = buffer->buffers[cpu];
3377                entries += rb_num_of_entries(cpu_buffer);
3378        }
3379
3380        return entries;
3381}
3382EXPORT_SYMBOL_GPL(ring_buffer_entries);
3383
3384/**
3385 * ring_buffer_overruns - get the number of overruns in buffer
3386 * @buffer: The ring buffer
3387 *
3388 * Returns the total number of overruns in the ring buffer
3389 * (all CPU entries)
3390 */
3391unsigned long ring_buffer_overruns(struct ring_buffer *buffer)
3392{
3393        struct ring_buffer_per_cpu *cpu_buffer;
3394        unsigned long overruns = 0;
3395        int cpu;
3396
3397        /* if you care about this being correct, lock the buffer */
3398        for_each_buffer_cpu(buffer, cpu) {
3399                cpu_buffer = buffer->buffers[cpu];
3400                overruns += local_read(&cpu_buffer->overrun);
3401        }
3402
3403        return overruns;
3404}
3405EXPORT_SYMBOL_GPL(ring_buffer_overruns);
3406
3407static void rb_iter_reset(struct ring_buffer_iter *iter)
3408{
3409        struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
3410
3411        /* Iterator usage is expected to have record disabled */
3412        iter->head_page = cpu_buffer->reader_page;
3413        iter->head = cpu_buffer->reader_page->read;
3414
3415        iter->cache_reader_page = iter->head_page;
3416        iter->cache_read = cpu_buffer->read;
3417
3418        if (iter->head)
3419                iter->read_stamp = cpu_buffer->read_stamp;
3420        else
3421                iter->read_stamp = iter->head_page->page->time_stamp;
3422}
3423
3424/**
3425 * ring_buffer_iter_reset - reset an iterator
3426 * @iter: The iterator to reset
3427 *
3428 * Resets the iterator, so that it will start from the beginning
3429 * again.
3430 */
3431void ring_buffer_iter_reset(struct ring_buffer_iter *iter)
3432{
3433        struct ring_buffer_per_cpu *cpu_buffer;
3434        unsigned long flags;
3435
3436        if (!iter)
3437                return;
3438
3439        cpu_buffer = iter->cpu_buffer;
3440
3441        raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
3442        rb_iter_reset(iter);
3443        raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
3444}
3445EXPORT_SYMBOL_GPL(ring_buffer_iter_reset);
3446
3447/**
3448 * ring_buffer_iter_empty - check if an iterator has no more to read
3449 * @iter: The iterator to check
3450 */
3451int ring_buffer_iter_empty(struct ring_buffer_iter *iter)
3452{
3453        struct ring_buffer_per_cpu *cpu_buffer;
3454
3455        cpu_buffer = iter->cpu_buffer;
3456
3457        return iter->head_page == cpu_buffer->commit_page &&
3458                iter->head == rb_commit_index(cpu_buffer);
3459}
3460EXPORT_SYMBOL_GPL(ring_buffer_iter_empty);
3461
3462static void
3463rb_update_read_stamp(struct ring_buffer_per_cpu *cpu_buffer,
3464                     struct ring_buffer_event *event)
3465{
3466        u64 delta;
3467
3468        switch (event->type_len) {
3469        case RINGBUF_TYPE_PADDING:
3470                return;
3471
3472        case RINGBUF_TYPE_TIME_EXTEND:
3473                delta = event->array[0];
3474                delta <<= TS_SHIFT;
3475                delta += event->time_delta;
3476                cpu_buffer->read_stamp += delta;
3477                return;
3478
3479        case RINGBUF_TYPE_TIME_STAMP:
3480                /* FIXME: not implemented */
3481                return;
3482
3483        case RINGBUF_TYPE_DATA:
3484                cpu_buffer->read_stamp += event->time_delta;
3485                return;
3486
3487        default:
3488                BUG();
3489        }
3490        return;
3491}
3492
3493static void
3494rb_update_iter_read_stamp(struct ring_buffer_iter *iter,
3495                          struct ring_buffer_event *event)
3496{
3497        u64 delta;
3498
3499        switch (event->type_len) {
3500        case RINGBUF_TYPE_PADDING:
3501                return;
3502
3503        case RINGBUF_TYPE_TIME_EXTEND:
3504                delta = event->array[0];
3505                delta <<= TS_SHIFT;
3506                delta += event->time_delta;
3507                iter->read_stamp += delta;
3508                return;
3509
3510        case RINGBUF_TYPE_TIME_STAMP:
3511                /* FIXME: not implemented */
3512                return;
3513
3514        case RINGBUF_TYPE_DATA:
3515                iter->read_stamp += event->time_delta;
3516                return;
3517
3518        default:
3519                BUG();
3520        }
3521        return;
3522}
3523
3524static struct buffer_page *
3525rb_get_reader_page(struct ring_buffer_per_cpu *cpu_buffer)
3526{
3527        struct buffer_page *reader = NULL;
3528        unsigned long overwrite;
3529        unsigned long flags;
3530        int nr_loops = 0;
3531        int ret;
3532
3533        local_irq_save(flags);
3534        arch_spin_lock(&cpu_buffer->lock);
3535
3536 again:
3537        /*
3538         * This should normally only loop twice. But because the
3539         * start of the reader inserts an empty page, it causes
3540         * a case where we will loop three times. There should be no
3541         * reason to loop four times (that I know of).
3542         */
3543        if (RB_WARN_ON(cpu_buffer, ++nr_loops > 3)) {
3544                reader = NULL;
3545                goto out;
3546        }
3547
3548        reader = cpu_buffer->reader_page;
3549
3550        /* If there's more to read, return this page */
3551        if (cpu_buffer->reader_page->read < rb_page_size(reader))
3552                goto out;
3553
3554        /* Never should we have an index greater than the size */
3555        if (RB_WARN_ON(cpu_buffer,
3556                       cpu_buffer->reader_page->read > rb_page_size(reader)))
3557                goto out;
3558
3559        /* check if we caught up to the tail */
3560        reader = NULL;
3561        if (cpu_buffer->commit_page == cpu_buffer->reader_page)
3562                goto out;
3563
3564        /* Don't bother swapping if the ring buffer is empty */
3565        if (rb_num_of_entries(cpu_buffer) == 0)
3566                goto out;
3567
3568        /*
3569         * Reset the reader page to size zero.
3570         */
3571        local_set(&cpu_buffer->reader_page->write, 0);
3572        local_set(&cpu_buffer->reader_page->entries, 0);
3573        local_set(&cpu_buffer->reader_page->page->commit, 0);
3574        cpu_buffer->reader_page->real_end = 0;
3575
3576 spin:
3577        /*
3578         * Splice the empty reader page into the list around the head.
3579         */
3580        reader = rb_set_head_page(cpu_buffer);
3581        if (!reader)
3582                goto out;
3583        cpu_buffer->reader_page->list.next = rb_list_head(reader->list.next);
3584        cpu_buffer->reader_page->list.prev = reader->list.prev;
3585
3586        /*
3587         * cpu_buffer->pages just needs to point to the buffer, it
3588         *  has no specific buffer page to point to. Lets move it out
3589         *  of our way so we don't accidentally swap it.
3590         */
3591        cpu_buffer->pages = reader->list.prev;
3592
3593        /* The reader page will be pointing to the new head */
3594        rb_set_list_to_head(cpu_buffer, &cpu_buffer->reader_page->list);
3595
3596        /*
3597         * We want to make sure we read the overruns after we set up our
3598         * pointers to the next object. The writer side does a
3599         * cmpxchg to cross pages which acts as the mb on the writer
3600         * side. Note, the reader will constantly fail the swap
3601         * while the writer is updating the pointers, so this
3602         * guarantees that the overwrite recorded here is the one we
3603         * want to compare with the last_overrun.
3604         */
3605        smp_mb();
3606        overwrite = local_read(&(cpu_buffer->overrun));
3607
3608        /*
3609         * Here's the tricky part.
3610         *
3611         * We need to move the pointer past the header page.
3612         * But we can only do that if a writer is not currently
3613         * moving it. The page before the header page has the
3614         * flag bit '1' set if it is pointing to the page we want.
3615         * but if the writer is in the process of moving it
3616         * than it will be '2' or already moved '0'.
3617         */
3618
3619        ret = rb_head_page_replace(reader, cpu_buffer->reader_page);
3620
3621        /*
3622         * If we did not convert it, then we must try again.
3623         */
3624        if (!ret)
3625                goto spin;
3626
3627        /*
3628         * Yeah! We succeeded in replacing the page.
3629         *
3630         * Now make the new head point back to the reader page.
3631         */
3632        rb_list_head(reader->list.next)->prev = &cpu_buffer->reader_page->list;
3633        rb_inc_page(cpu_buffer, &cpu_buffer->head_page);
3634
3635        /* Finally update the reader page to the new head */
3636        cpu_buffer->reader_page = reader;
3637        rb_reset_reader_page(cpu_buffer);
3638
3639        if (overwrite != cpu_buffer->last_overrun) {
3640                cpu_buffer->lost_events = overwrite - cpu_buffer->last_overrun;
3641                cpu_buffer->last_overrun = overwrite;
3642        }
3643
3644        goto again;
3645
3646 out:
3647        arch_spin_unlock(&cpu_buffer->lock);
3648        local_irq_restore(flags);
3649
3650        return reader;
3651}
3652
3653static void rb_advance_reader(struct ring_buffer_per_cpu *cpu_buffer)
3654{
3655        struct ring_buffer_event *event;
3656        struct buffer_page *reader;
3657        unsigned length;
3658
3659        reader = rb_get_reader_page(cpu_buffer);
3660
3661        /* This function should not be called when buffer is empty */
3662        if (RB_WARN_ON(cpu_buffer, !reader))
3663                return;
3664
3665        event = rb_reader_event(cpu_buffer);
3666
3667        if (event->type_len <= RINGBUF_TYPE_DATA_TYPE_LEN_MAX)
3668                cpu_buffer->read++;
3669
3670        rb_update_read_stamp(cpu_buffer, event);
3671
3672        length = rb_event_length(event);
3673        cpu_buffer->reader_page->read += length;
3674}
3675
3676static void rb_advance_iter(struct ring_buffer_iter *iter)
3677{
3678        struct ring_buffer_per_cpu *cpu_buffer;
3679        struct ring_buffer_event *event;
3680        unsigned length;
3681
3682        cpu_buffer = iter->cpu_buffer;
3683
3684        /*
3685         * Check if we are at the end of the buffer.
3686         */
3687        if (iter->head >= rb_page_size(iter->head_page)) {
3688                /* discarded commits can make the page empty */
3689                if (iter->head_page == cpu_buffer->commit_page)
3690                        return;
3691                rb_inc_iter(iter);
3692                return;
3693        }
3694
3695        event = rb_iter_head_event(iter);
3696
3697        length = rb_event_length(event);
3698
3699        /*
3700         * This should not be called to advance the header if we are
3701         * at the tail of the buffer.
3702         */
3703        if (RB_WARN_ON(cpu_buffer,
3704                       (iter->head_page == cpu_buffer->commit_page) &&
3705                       (iter->head + length > rb_commit_index(cpu_buffer))))
3706                return;
3707
3708        rb_update_iter_read_stamp(iter, event);
3709
3710        iter->head += length;
3711
3712        /* check for end of page padding */
3713        if ((iter->head >= rb_page_size(iter->head_page)) &&
3714            (iter->head_page != cpu_buffer->commit_page))
3715                rb_inc_iter(iter);
3716}
3717
3718static int rb_lost_events(struct ring_buffer_per_cpu *cpu_buffer)
3719{
3720        return cpu_buffer->lost_events;
3721}
3722
3723static struct ring_buffer_event *
3724rb_buffer_peek(struct ring_buffer_per_cpu *cpu_buffer, u64 *ts,
3725               unsigned long *lost_events)
3726{
3727        struct ring_buffer_event *event;
3728        struct buffer_page *reader;
3729        int nr_loops = 0;
3730
3731 again:
3732        /*
3733         * We repeat when a time extend is encountered.
3734         * Since the time extend is always attached to a data event,
3735         * we should never loop more than once.
3736         * (We never hit the following condition more than twice).
3737         */
3738        if (RB_WARN_ON(cpu_buffer, ++nr_loops > 2))
3739                return NULL;
3740
3741        reader = rb_get_reader_page(cpu_buffer);
3742        if (!reader)
3743                return NULL;
3744
3745        event = rb_reader_event(cpu_buffer);
3746
3747        switch (event->type_len) {
3748        case RINGBUF_TYPE_PADDING:
3749                if (rb_null_event(event))
3750                        RB_WARN_ON(cpu_buffer, 1);
3751                /*
3752                 * Because the writer could be discarding every
3753                 * event it creates (which would probably be bad)
3754                 * if we were to go back to "again" then we may never
3755                 * catch up, and will trigger the warn on, or lock
3756                 * the box. Return the padding, and we will release
3757                 * the current locks, and try again.
3758                 */
3759                return event;
3760
3761        case RINGBUF_TYPE_TIME_EXTEND:
3762                /* Internal data, OK to advance */
3763                rb_advance_reader(cpu_buffer);
3764                goto again;
3765
3766        case RINGBUF_TYPE_TIME_STAMP:
3767                /* FIXME: not implemented */
3768                rb_advance_reader(cpu_buffer);
3769                goto again;
3770
3771        case RINGBUF_TYPE_DATA:
3772                if (ts) {
3773                        *ts = cpu_buffer->read_stamp + event->time_delta;
3774                        ring_buffer_normalize_time_stamp(cpu_buffer->buffer,
3775                                                         cpu_buffer->cpu, ts);
3776                }
3777                if (lost_events)
3778                        *lost_events = rb_lost_events(cpu_buffer);
3779                return event;
3780
3781        default:
3782                BUG();
3783        }
3784
3785        return NULL;
3786}
3787EXPORT_SYMBOL_GPL(ring_buffer_peek);
3788
3789static struct ring_buffer_event *
3790rb_iter_peek(struct ring_buffer_iter *iter, u64 *ts)
3791{
3792        struct ring_buffer *buffer;
3793        struct ring_buffer_per_cpu *cpu_buffer;
3794        struct ring_buffer_event *event;
3795        int nr_loops = 0;
3796
3797        cpu_buffer = iter->cpu_buffer;
3798        buffer = cpu_buffer->buffer;
3799
3800        /*
3801         * Check if someone performed a consuming read to
3802         * the buffer. A consuming read invalidates the iterator
3803         * and we need to reset the iterator in this case.
3804         */
3805        if (unlikely(iter->cache_read != cpu_buffer->read ||
3806                     iter->cache_reader_page != cpu_buffer->reader_page))
3807                rb_iter_reset(iter);
3808
3809 again:
3810        if (ring_buffer_iter_empty(iter))
3811                return NULL;
3812
3813        /*
3814         * We repeat when a time extend is encountered or we hit
3815         * the end of the page. Since the time extend is always attached
3816         * to a data event, we should never loop more than three times.
3817         * Once for going to next page, once on time extend, and
3818         * finally once to get the event.
3819         * (We never hit the following condition more than thrice).
3820         */
3821        if (RB_WARN_ON(cpu_buffer, ++nr_loops > 3))
3822                return NULL;
3823
3824        if (rb_per_cpu_empty(cpu_buffer))
3825                return NULL;
3826
3827        if (iter->head >= rb_page_size(iter->head_page)) {
3828                rb_inc_iter(iter);
3829                goto again;
3830        }
3831
3832        event = rb_iter_head_event(iter);
3833
3834        switch (event->type_len) {
3835        case RINGBUF_TYPE_PADDING:
3836                if (rb_null_event(event)) {
3837                        rb_inc_iter(iter);
3838                        goto again;
3839                }
3840                rb_advance_iter(iter);
3841                return event;
3842
3843        case RINGBUF_TYPE_TIME_EXTEND:
3844                /* Internal data, OK to advance */
3845                rb_advance_iter(iter);
3846                goto again;
3847
3848        case RINGBUF_TYPE_TIME_STAMP:
3849                /* FIXME: not implemented */
3850                rb_advance_iter(iter);
3851                goto again;
3852
3853        case RINGBUF_TYPE_DATA:
3854                if (ts) {
3855                        *ts = iter->read_stamp + event->time_delta;
3856                        ring_buffer_normalize_time_stamp(buffer,
3857                                                         cpu_buffer->cpu, ts);
3858                }
3859                return event;
3860
3861        default:
3862                BUG();
3863        }
3864
3865        return NULL;
3866}
3867EXPORT_SYMBOL_GPL(ring_buffer_iter_peek);
3868
3869static inline int rb_ok_to_lock(void)
3870{
3871        /*
3872         * If an NMI die dumps out the content of the ring buffer
3873         * do not grab locks. We also permanently disable the ring
3874         * buffer too. A one time deal is all you get from reading
3875         * the ring buffer from an NMI.
3876         */
3877        if (likely(!in_nmi()))
3878                return 1;
3879
3880        tracing_off_permanent();
3881        return 0;
3882}
3883
3884/**
3885 * ring_buffer_peek - peek at the next event to be read
3886 * @buffer: The ring buffer to read
3887 * @cpu: The cpu to peak at
3888 * @ts: The timestamp counter of this event.
3889 * @lost_events: a variable to store if events were lost (may be NULL)
3890 *
3891 * This will return the event that will be read next, but does
3892 * not consume the data.
3893 */
3894struct ring_buffer_event *
3895ring_buffer_peek(struct ring_buffer *buffer, int cpu, u64 *ts,
3896                 unsigned long *lost_events)
3897{
3898        struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
3899        struct ring_buffer_event *event;
3900        unsigned long flags;
3901        int dolock;
3902
3903        if (!cpumask_test_cpu(cpu, buffer->cpumask))
3904                return NULL;
3905
3906        dolock = rb_ok_to_lock();
3907 again:
3908        local_irq_save(flags);
3909        if (dolock)
3910                raw_spin_lock(&cpu_buffer->reader_lock);
3911        event = rb_buffer_peek(cpu_buffer, ts, lost_events);
3912        if (event && event->type_len == RINGBUF_TYPE_PADDING)
3913                rb_advance_reader(cpu_buffer);
3914        if (dolock)
3915                raw_spin_unlock(&cpu_buffer->reader_lock);
3916        local_irq_restore(flags);
3917
3918        if (event && event->type_len == RINGBUF_TYPE_PADDING)
3919                goto again;
3920
3921        return event;
3922}
3923
3924/**
3925 * ring_buffer_iter_peek - peek at the next event to be read
3926 * @iter: The ring buffer iterator
3927 * @ts: The timestamp counter of this event.
3928 *
3929 * This will return the event that will be read next, but does
3930 * not increment the iterator.
3931 */
3932struct ring_buffer_event *
3933ring_buffer_iter_peek(struct ring_buffer_iter *iter, u64 *ts)
3934{
3935        struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
3936        struct ring_buffer_event *event;
3937        unsigned long flags;
3938
3939 again:
3940        raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
3941        event = rb_iter_peek(iter, ts);
3942        raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
3943
3944        if (event && event->type_len == RINGBUF_TYPE_PADDING)
3945                goto again;
3946
3947        return event;
3948}
3949
3950/**
3951 * ring_buffer_consume - return an event and consume it
3952 * @buffer: The ring buffer to get the next event from
3953 * @cpu: the cpu to read the buffer from
3954 * @ts: a variable to store the timestamp (may be NULL)
3955 * @lost_events: a variable to store if events were lost (may be NULL)
3956 *
3957 * Returns the next event in the ring buffer, and that event is consumed.
3958 * Meaning, that sequential reads will keep returning a different event,
3959 * and eventually empty the ring buffer if the producer is slower.
3960 */
3961struct ring_buffer_event *
3962ring_buffer_consume(struct ring_buffer *buffer, int cpu, u64 *ts,
3963                    unsigned long *lost_events)
3964{
3965        struct ring_buffer_per_cpu *cpu_buffer;
3966        struct ring_buffer_event *event = NULL;
3967        unsigned long flags;
3968        int dolock;
3969
3970        dolock = rb_ok_to_lock();
3971
3972 again:
3973        /* might be called in atomic */
3974        preempt_disable();
3975
3976        if (!cpumask_test_cpu(cpu, buffer->cpumask))
3977                goto out;
3978
3979        cpu_buffer = buffer->buffers[cpu];
3980        local_irq_save(flags);
3981        if (dolock)
3982                raw_spin_lock(&cpu_buffer->reader_lock);
3983
3984        event = rb_buffer_peek(cpu_buffer, ts, lost_events);
3985        if (event) {
3986                cpu_buffer->lost_events = 0;
3987                rb_advance_reader(cpu_buffer);
3988        }
3989
3990        if (dolock)
3991                raw_spin_unlock(&cpu_buffer->reader_lock);
3992        local_irq_restore(flags);
3993
3994 out:
3995        preempt_enable();
3996
3997        if (event && event->type_len == RINGBUF_TYPE_PADDING)
3998                goto again;
3999
4000        return event;
4001}
4002EXPORT_SYMBOL_GPL(ring_buffer_consume);
4003
4004/**
4005 * ring_buffer_read_prepare - Prepare for a non consuming read of the buffer
4006 * @buffer: The ring buffer to read from
4007 * @cpu: The cpu buffer to iterate over
4008 *
4009 * This performs the initial preparations necessary to iterate
4010 * through the buffer.  Memory is allocated, buffer recording
4011 * is disabled, and the iterator pointer is returned to the caller.
4012 *
4013 * Disabling buffer recordng prevents the reading from being
4014 * corrupted. This is not a consuming read, so a producer is not
4015 * expected.
4016 *
4017 * After a sequence of ring_buffer_read_prepare calls, the user is
4018 * expected to make at least one call to ring_buffer_read_prepare_sync.
4019 * Afterwards, ring_buffer_read_start is invoked to get things going
4020 * for real.
4021 *
4022 * This overall must be paired with ring_buffer_read_finish.
4023 */
4024struct ring_buffer_iter *
4025ring_buffer_read_prepare(struct ring_buffer *buffer, int cpu)
4026{
4027        struct ring_buffer_per_cpu *cpu_buffer;
4028        struct ring_buffer_iter *iter;
4029
4030        if (!cpumask_test_cpu(cpu, buffer->cpumask))
4031                return NULL;
4032
4033        iter = kmalloc(sizeof(*iter), GFP_KERNEL);
4034        if (!iter)
4035                return NULL;
4036
4037        cpu_buffer = buffer->buffers[cpu];
4038
4039        iter->cpu_buffer = cpu_buffer;
4040
4041        atomic_inc(&buffer->resize_disabled);
4042        atomic_inc(&cpu_buffer->record_disabled);
4043
4044        return iter;
4045}
4046EXPORT_SYMBOL_GPL(ring_buffer_read_prepare);
4047
4048/**
4049 * ring_buffer_read_prepare_sync - Synchronize a set of prepare calls
4050 *
4051 * All previously invoked ring_buffer_read_prepare calls to prepare
4052 * iterators will be synchronized.  Afterwards, read_buffer_read_start
4053 * calls on those iterators are allowed.
4054 */
4055void
4056ring_buffer_read_prepare_sync(void)
4057{
4058        synchronize_sched();
4059}
4060EXPORT_SYMBOL_GPL(ring_buffer_read_prepare_sync);
4061
4062/**
4063 * ring_buffer_read_start - start a non consuming read of the buffer
4064 * @iter: The iterator returned by ring_buffer_read_prepare
4065 *
4066 * This finalizes the startup of an iteration through the buffer.
4067 * The iterator comes from a call to ring_buffer_read_prepare and
4068 * an intervening ring_buffer_read_prepare_sync must have been
4069 * performed.
4070 *
4071 * Must be paired with ring_buffer_read_finish.
4072 */
4073void
4074ring_buffer_read_start(struct ring_buffer_iter *iter)
4075{
4076        struct ring_buffer_per_cpu *cpu_buffer;
4077        unsigned long flags;
4078
4079        if (!iter)
4080                return;
4081
4082        cpu_buffer = iter->cpu_buffer;
4083
4084        raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
4085        arch_spin_lock(&cpu_buffer->lock);
4086        rb_iter_reset(iter);
4087        arch_spin_unlock(&cpu_buffer->lock);
4088        raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
4089}
4090EXPORT_SYMBOL_GPL(ring_buffer_read_start);
4091
4092/**
4093 * ring_buffer_read_finish - finish reading the iterator of the buffer
4094 * @iter: The iterator retrieved by ring_buffer_start
4095 *
4096 * This re-enables the recording to the buffer, and frees the
4097 * iterator.
4098 */
4099void
4100ring_buffer_read_finish(struct ring_buffer_iter *iter)
4101{
4102        struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
4103        unsigned long flags;
4104
4105        /*
4106         * Ring buffer is disabled from recording, here's a good place
4107         * to check the integrity of the ring buffer.
4108         * Must prevent readers from trying to read, as the check
4109         * clears the HEAD page and readers require it.
4110         */
4111        raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
4112        rb_check_pages(cpu_buffer);
4113        raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
4114
4115        atomic_dec(&cpu_buffer->record_disabled);
4116        atomic_dec(&cpu_buffer->buffer->resize_disabled);
4117        kfree(iter);
4118}
4119EXPORT_SYMBOL_GPL(ring_buffer_read_finish);
4120
4121/**
4122 * ring_buffer_read - read the next item in the ring buffer by the iterator
4123 * @iter: The ring buffer iterator
4124 * @ts: The time stamp of the event read.
4125 *
4126 * This reads the next event in the ring buffer and increments the iterator.
4127 */
4128struct ring_buffer_event *
4129ring_buffer_read(struct ring_buffer_iter *iter, u64 *ts)
4130{
4131        struct ring_buffer_event *event;
4132        struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
4133        unsigned long flags;
4134
4135        raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
4136 again:
4137        event = rb_iter_peek(iter, ts);
4138        if (!event)
4139                goto out;
4140
4141        if (event->type_len == RINGBUF_TYPE_PADDING)
4142                goto again;
4143
4144        rb_advance_iter(iter);
4145 out:
4146        raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
4147
4148        return event;
4149}
4150EXPORT_SYMBOL_GPL(ring_buffer_read);
4151
4152/**
4153 * ring_buffer_size - return the size of the ring buffer (in bytes)
4154 * @buffer: The ring buffer.
4155 */
4156unsigned long ring_buffer_size(struct ring_buffer *buffer, int cpu)
4157{
4158        /*
4159         * Earlier, this method returned
4160         *      BUF_PAGE_SIZE * buffer->nr_pages
4161         * Since the nr_pages field is now removed, we have converted this to
4162         * return the per cpu buffer value.
4163         */
4164        if (!cpumask_test_cpu(cpu, buffer->cpumask))
4165                return 0;
4166
4167        return BUF_PAGE_SIZE * buffer->buffers[cpu]->nr_pages;
4168}
4169EXPORT_SYMBOL_GPL(ring_buffer_size);
4170
4171static void
4172rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer)
4173{
4174        rb_head_page_deactivate(cpu_buffer);
4175
4176        cpu_buffer->head_page
4177                = list_entry(cpu_buffer->pages, struct buffer_page, list);
4178        local_set(&cpu_buffer->head_page->write, 0);
4179        local_set(&cpu_buffer->head_page->entries, 0);
4180        local_set(&cpu_buffer->head_page->page->commit, 0);
4181
4182        cpu_buffer->head_page->read = 0;
4183
4184        cpu_buffer->tail_page = cpu_buffer->head_page;
4185        cpu_buffer->commit_page = cpu_buffer->head_page;
4186
4187        INIT_LIST_HEAD(&cpu_buffer->reader_page->list);
4188        INIT_LIST_HEAD(&cpu_buffer->new_pages);
4189        local_set(&cpu_buffer->reader_page->write, 0);
4190        local_set(&cpu_buffer->reader_page->entries, 0);
4191        local_set(&cpu_buffer->reader_page->page->commit, 0);
4192        cpu_buffer->reader_page->read = 0;
4193
4194        local_set(&cpu_buffer->entries_bytes, 0);
4195        local_set(&cpu_buffer->overrun, 0);
4196        local_set(&cpu_buffer->commit_overrun, 0);
4197        local_set(&cpu_buffer->dropped_events, 0);
4198        local_set(&cpu_buffer->entries, 0);
4199        local_set(&cpu_buffer->committing, 0);
4200        local_set(&cpu_buffer->commits, 0);
4201        cpu_buffer->read = 0;
4202        cpu_buffer->read_bytes = 0;
4203
4204        cpu_buffer->write_stamp = 0;
4205        cpu_buffer->read_stamp = 0;
4206
4207        cpu_buffer->lost_events = 0;
4208        cpu_buffer->last_overrun = 0;
4209
4210        rb_head_page_activate(cpu_buffer);
4211}
4212
4213/**
4214 * ring_buffer_reset_cpu - reset a ring buffer per CPU buffer
4215 * @buffer: The ring buffer to reset a per cpu buffer of
4216 * @cpu: The CPU buffer to be reset
4217 */
4218void ring_buffer_reset_cpu(struct ring_buffer *buffer, int cpu)
4219{
4220        struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
4221        unsigned long flags;
4222
4223        if (!cpumask_test_cpu(cpu, buffer->cpumask))
4224                return;
4225
4226        atomic_inc(&buffer->resize_disabled);
4227        atomic_inc(&cpu_buffer->record_disabled);
4228
4229        /* Make sure all commits have finished */
4230        synchronize_sched();
4231
4232        raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
4233
4234        if (RB_WARN_ON(cpu_buffer, local_read(&cpu_buffer->committing)))
4235                goto out;
4236
4237        arch_spin_lock(&cpu_buffer->lock);
4238
4239        rb_reset_cpu(cpu_buffer);
4240
4241        arch_spin_unlock(&cpu_buffer->lock);
4242
4243 out:
4244        raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
4245
4246        atomic_dec(&cpu_buffer->record_disabled);
4247        atomic_dec(&buffer->resize_disabled);
4248}
4249EXPORT_SYMBOL_GPL(ring_buffer_reset_cpu);
4250
4251/**
4252 * ring_buffer_reset - reset a ring buffer
4253 * @buffer: The ring buffer to reset all cpu buffers
4254 */
4255void ring_buffer_reset(struct ring_buffer *buffer)
4256{
4257        int cpu;
4258
4259        for_each_buffer_cpu(buffer, cpu)
4260                ring_buffer_reset_cpu(buffer, cpu);
4261}
4262EXPORT_SYMBOL_GPL(ring_buffer_reset);
4263
4264/**
4265 * rind_buffer_empty - is the ring buffer empty?
4266 * @buffer: The ring buffer to test
4267 */
4268int ring_buffer_empty(struct ring_buffer *buffer)
4269{
4270        struct ring_buffer_per_cpu *cpu_buffer;
4271        unsigned long flags;
4272        int dolock;
4273        int cpu;
4274        int ret;
4275
4276        dolock = rb_ok_to_lock();
4277
4278        /* yes this is racy, but if you don't like the race, lock the buffer */
4279        for_each_buffer_cpu(buffer, cpu) {
4280                cpu_buffer = buffer->buffers[cpu];
4281                local_irq_save(flags);
4282                if (dolock)
4283                        raw_spin_lock(&cpu_buffer->reader_lock);
4284                ret = rb_per_cpu_empty(cpu_buffer);
4285                if (dolock)
4286                        raw_spin_unlock(&cpu_buffer->reader_lock);
4287                local_irq_restore(flags);
4288
4289                if (!ret)
4290                        return 0;
4291        }
4292
4293        return 1;
4294}
4295EXPORT_SYMBOL_GPL(ring_buffer_empty);
4296
4297/**
4298 * ring_buffer_empty_cpu - is a cpu buffer of a ring buffer empty?
4299 * @buffer: The ring buffer
4300 * @cpu: The CPU buffer to test
4301 */
4302int ring_buffer_empty_cpu(struct ring_buffer *buffer, int cpu)
4303{
4304        struct ring_buffer_per_cpu *cpu_buffer;
4305        unsigned long flags;
4306        int dolock;
4307        int ret;
4308
4309        if (!cpumask_test_cpu(cpu, buffer->cpumask))
4310                return 1;
4311
4312        dolock = rb_ok_to_lock();
4313
4314        cpu_buffer = buffer->buffers[cpu];
4315        local_irq_save(flags);
4316        if (dolock)
4317                raw_spin_lock(&cpu_buffer->reader_lock);
4318        ret = rb_per_cpu_empty(cpu_buffer);
4319        if (dolock)
4320                raw_spin_unlock(&cpu_buffer->reader_lock);
4321        local_irq_restore(flags);
4322
4323        return ret;
4324}
4325EXPORT_SYMBOL_GPL(ring_buffer_empty_cpu);
4326
4327#ifdef CONFIG_RING_BUFFER_ALLOW_SWAP
4328/**
4329 * ring_buffer_swap_cpu - swap a CPU buffer between two ring buffers
4330 * @buffer_a: One buffer to swap with
4331 * @buffer_b: The other buffer to swap with
4332 *
4333 * This function is useful for tracers that want to take a "snapshot"
4334 * of a CPU buffer and has another back up buffer lying around.
4335 * it is expected that the tracer handles the cpu buffer not being
4336 * used at the moment.
4337 */
4338int ring_buffer_swap_cpu(struct ring_buffer *buffer_a,
4339                         struct ring_buffer *buffer_b, int cpu)
4340{
4341        struct ring_buffer_per_cpu *cpu_buffer_a;
4342        struct ring_buffer_per_cpu *cpu_buffer_b;
4343        int ret = -EINVAL;
4344
4345        if (!cpumask_test_cpu(cpu, buffer_a->cpumask) ||
4346            !cpumask_test_cpu(cpu, buffer_b->cpumask))
4347                goto out;
4348
4349        cpu_buffer_a = buffer_a->buffers[cpu];
4350        cpu_buffer_b = buffer_b->buffers[cpu];
4351
4352        /* At least make sure the two buffers are somewhat the same */
4353        if (cpu_buffer_a->nr_pages != cpu_buffer_b->nr_pages)
4354                goto out;
4355
4356        ret = -EAGAIN;
4357
4358        if (ring_buffer_flags != RB_BUFFERS_ON)
4359                goto out;
4360
4361        if (atomic_read(&buffer_a->record_disabled))
4362                goto out;
4363
4364        if (atomic_read(&buffer_b->record_disabled))
4365                goto out;
4366
4367        if (atomic_read(&cpu_buffer_a->record_disabled))
4368                goto out;
4369
4370        if (atomic_read(&cpu_buffer_b->record_disabled))
4371                goto out;
4372
4373        /*
4374         * We can't do a synchronize_sched here because this
4375         * function can be called in atomic context.
4376         * Normally this will be called from the same CPU as cpu.
4377         * If not it's up to the caller to protect this.
4378         */
4379        atomic_inc(&cpu_buffer_a->record_disabled);
4380        atomic_inc(&cpu_buffer_b->record_disabled);
4381
4382        ret = -EBUSY;
4383        if (local_read(&cpu_buffer_a->committing))
4384                goto out_dec;
4385        if (local_read(&cpu_buffer_b->committing))
4386                goto out_dec;
4387
4388        buffer_a->buffers[cpu] = cpu_buffer_b;
4389        buffer_b->buffers[cpu] = cpu_buffer_a;
4390
4391        cpu_buffer_b->buffer = buffer_a;
4392        cpu_buffer_a->buffer = buffer_b;
4393
4394        ret = 0;
4395
4396out_dec:
4397        atomic_dec(&cpu_buffer_a->record_disabled);
4398        atomic_dec(&cpu_buffer_b->record_disabled);
4399out:
4400        return ret;
4401}
4402EXPORT_SYMBOL_GPL(ring_buffer_swap_cpu);
4403#endif /* CONFIG_RING_BUFFER_ALLOW_SWAP */
4404
4405/**
4406 * ring_buffer_alloc_read_page - allocate a page to read from buffer
4407 * @buffer: the buffer to allocate for.
4408 * @cpu: the cpu buffer to allocate.
4409 *
4410 * This function is used in conjunction with ring_buffer_read_page.
4411 * When reading a full page from the ring buffer, these functions
4412 * can be used to speed up the process. The calling function should
4413 * allocate a few pages first with this function. Then when it
4414 * needs to get pages from the ring buffer, it passes the result
4415 * of this function into ring_buffer_read_page, which will swap
4416 * the page that was allocated, with the read page of the buffer.
4417 *
4418 * Returns:
4419 *  The page allocated, or NULL on error.
4420 */
4421void *ring_buffer_alloc_read_page(struct ring_buffer *buffer, int cpu)
4422{
4423        struct buffer_data_page *bpage;
4424        struct page *page;
4425
4426        page = alloc_pages_node(cpu_to_node(cpu),
4427                                GFP_KERNEL | __GFP_NORETRY, 0);
4428        if (!page)
4429                return NULL;
4430
4431        bpage = page_address(page);
4432
4433        rb_init_page(bpage);
4434
4435        return bpage;
4436}
4437EXPORT_SYMBOL_GPL(ring_buffer_alloc_read_page);
4438
4439/**
4440 * ring_buffer_free_read_page - free an allocated read page
4441 * @buffer: the buffer the page was allocate for
4442 * @data: the page to free
4443 *
4444 * Free a page allocated from ring_buffer_alloc_read_page.
4445 */
4446void ring_buffer_free_read_page(struct ring_buffer *buffer, void *data)
4447{
4448        free_page((unsigned long)data);
4449}
4450EXPORT_SYMBOL_GPL(ring_buffer_free_read_page);
4451
4452/**
4453 * ring_buffer_read_page - extract a page from the ring buffer
4454 * @buffer: buffer to extract from
4455 * @data_page: the page to use allocated from ring_buffer_alloc_read_page
4456 * @len: amount to extract
4457 * @cpu: the cpu of the buffer to extract
4458 * @full: should the extraction only happen when the page is full.
4459 *
4460 * This function will pull out a page from the ring buffer and consume it.
4461 * @data_page must be the address of the variable that was returned
4462 * from ring_buffer_alloc_read_page. This is because the page might be used
4463 * to swap with a page in the ring buffer.
4464 *
4465 * for example:
4466 *      rpage = ring_buffer_alloc_read_page(buffer, cpu);
4467 *      if (!rpage)
4468 *              return error;
4469 *      ret = ring_buffer_read_page(buffer, &rpage, len, cpu, 0);
4470 *      if (ret >= 0)
4471 *              process_page(rpage, ret);
4472 *
4473 * When @full is set, the function will not return true unless
4474 * the writer is off the reader page.
4475 *
4476 * Note: it is up to the calling functions to handle sleeps and wakeups.
4477 *  The ring buffer can be used anywhere in the kernel and can not
4478 *  blindly call wake_up. The layer that uses the ring buffer must be
4479 *  responsible for that.
4480 *
4481 * Returns:
4482 *  >=0 if data has been transferred, returns the offset of consumed data.
4483 *  <0 if no data has been transferred.
4484 */
4485int ring_buffer_read_page(struct ring_buffer *buffer,
4486                          void **data_page, size_t len, int cpu, int full)
4487{
4488        struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
4489        struct ring_buffer_event *event;
4490        struct buffer_data_page *bpage;
4491        struct buffer_page *reader;
4492        unsigned long missed_events;
4493        unsigned long flags;
4494        unsigned int commit;
4495        unsigned int read;
4496        u64 save_timestamp;
4497        int ret = -1;
4498
4499        if (!cpumask_test_cpu(cpu, buffer->cpumask))
4500                goto out;
4501
4502        /*
4503         * If len is not big enough to hold the page header, then
4504         * we can not copy anything.
4505         */
4506        if (len <= BUF_PAGE_HDR_SIZE)
4507                goto out;
4508
4509        len -= BUF_PAGE_HDR_SIZE;
4510
4511        if (!data_page)
4512                goto out;
4513
4514        bpage = *data_page;
4515        if (!bpage)
4516                goto out;
4517
4518        raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
4519
4520        reader = rb_get_reader_page(cpu_buffer);
4521        if (!reader)
4522                goto out_unlock;
4523
4524        event = rb_reader_event(cpu_buffer);
4525
4526        read = reader->read;
4527        commit = rb_page_commit(reader);
4528
4529        /* Check if any events were dropped */
4530        missed_events = cpu_buffer->lost_events;
4531
4532        /*
4533         * If this page has been partially read or
4534         * if len is not big enough to read the rest of the page or
4535         * a writer is still on the page, then
4536         * we must copy the data from the page to the buffer.
4537         * Otherwise, we can simply swap the page with the one passed in.
4538         */
4539        if (read || (len < (commit - read)) ||
4540            cpu_buffer->reader_page == cpu_buffer->commit_page) {
4541                struct buffer_data_page *rpage = cpu_buffer->reader_page->page;
4542                unsigned int rpos = read;
4543                unsigned int pos = 0;
4544                unsigned int size;
4545
4546                if (full)
4547                        goto out_unlock;
4548
4549                if (len > (commit - read))
4550                        len = (commit - read);
4551
4552                /* Always keep the time extend and data together */
4553                size = rb_event_ts_length(event);
4554
4555                if (len < size)
4556                        goto out_unlock;
4557
4558                /* save the current timestamp, since the user will need it */
4559                save_timestamp = cpu_buffer->read_stamp;
4560
4561                /* Need to copy one event at a time */
4562                do {
4563                        /* We need the size of one event, because
4564                         * rb_advance_reader only advances by one event,
4565                         * whereas rb_event_ts_length may include the size of
4566                         * one or two events.
4567                         * We have already ensured there's enough space if this
4568                         * is a time extend. */
4569                        size = rb_event_length(event);
4570                        memcpy(bpage->data + pos, rpage->data + rpos, size);
4571
4572                        len -= size;
4573
4574                        rb_advance_reader(cpu_buffer);
4575                        rpos = reader->read;
4576                        pos += size;
4577
4578                        if (rpos >= commit)
4579                                break;
4580
4581                        event = rb_reader_event(cpu_buffer);
4582                        /* Always keep the time extend and data together */
4583                        size = rb_event_ts_length(event);
4584                } while (len >= size);
4585
4586                /* update bpage */
4587                local_set(&bpage->commit, pos);
4588                bpage->time_stamp = save_timestamp;
4589
4590                /* we copied everything to the beginning */
4591                read = 0;
4592        } else {
4593                /* update the entry counter */
4594                cpu_buffer->read += rb_page_entries(reader);
4595                cpu_buffer->read_bytes += BUF_PAGE_SIZE;
4596
4597                /* swap the pages */
4598                rb_init_page(bpage);
4599                bpage = reader->page;
4600                reader->page = *data_page;
4601                local_set(&reader->write, 0);
4602                local_set(&reader->entries, 0);
4603                reader->read = 0;
4604                *data_page = bpage;
4605
4606                /*
4607                 * Use the real_end for the data size,
4608                 * This gives us a chance to store the lost events
4609                 * on the page.
4610                 */
4611                if (reader->real_end)
4612                        local_set(&bpage->commit, reader->real_end);
4613        }
4614        ret = read;
4615
4616        cpu_buffer->lost_events = 0;
4617
4618        commit = local_read(&bpage->commit);
4619        /*
4620         * Set a flag in the commit field if we lost events
4621         */
4622        if (missed_events) {
4623                /* If there is room at the end of the page to save the
4624                 * missed events, then record it there.
4625                 */
4626                if (BUF_PAGE_SIZE - commit >= sizeof(missed_events)) {
4627                        memcpy(&bpage->data[commit], &missed_events,
4628                               sizeof(missed_events));
4629                        local_add(RB_MISSED_STORED, &bpage->commit);
4630                        commit += sizeof(missed_events);
4631                }
4632                local_add(RB_MISSED_EVENTS, &bpage->commit);
4633        }
4634
4635        /*
4636         * This page may be off to user land. Zero it out here.
4637         */
4638        if (commit < BUF_PAGE_SIZE)
4639                memset(&bpage->data[commit], 0, BUF_PAGE_SIZE - commit);
4640
4641 out_unlock:
4642        raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
4643
4644 out:
4645        return ret;
4646}
4647EXPORT_SYMBOL_GPL(ring_buffer_read_page);
4648
4649#ifdef CONFIG_HOTPLUG_CPU
4650static int rb_cpu_notify(struct notifier_block *self,
4651                         unsigned long action, void *hcpu)
4652{
4653        struct ring_buffer *buffer =
4654                container_of(self, struct ring_buffer, cpu_notify);
4655        long cpu = (long)hcpu;
4656        long nr_pages_same;
4657        int cpu_i;
4658        unsigned long nr_pages;
4659
4660        switch (action) {
4661        case CPU_UP_PREPARE:
4662        case CPU_UP_PREPARE_FROZEN:
4663                if (cpumask_test_cpu(cpu, buffer->cpumask))
4664                        return NOTIFY_OK;
4665
4666                nr_pages = 0;
4667                nr_pages_same = 1;
4668                /* check if all cpu sizes are same */
4669                for_each_buffer_cpu(buffer, cpu_i) {
4670                        /* fill in the size from first enabled cpu */
4671                        if (nr_pages == 0)
4672                                nr_pages = buffer->buffers[cpu_i]->nr_pages;
4673                        if (nr_pages != buffer->buffers[cpu_i]->nr_pages) {
4674                                nr_pages_same = 0;
4675                                break;
4676                        }
4677                }
4678                /* allocate minimum pages, user can later expand it */
4679                if (!nr_pages_same)
4680                        nr_pages = 2;
4681                buffer->buffers[cpu] =
4682                        rb_allocate_cpu_buffer(buffer, nr_pages, cpu);
4683                if (!buffer->buffers[cpu]) {
4684                        WARN(1, "failed to allocate ring buffer on CPU %ld\n",
4685                             cpu);
4686                        return NOTIFY_OK;
4687                }
4688                smp_wmb();
4689                cpumask_set_cpu(cpu, buffer->cpumask);
4690                break;
4691        case CPU_DOWN_PREPARE:
4692        case CPU_DOWN_PREPARE_FROZEN:
4693                /*
4694                 * Do nothing.
4695                 *  If we were to free the buffer, then the user would
4696                 *  lose any trace that was in the buffer.
4697                 */
4698                break;
4699        default:
4700                break;
4701        }
4702        return NOTIFY_OK;
4703}
4704#endif
4705
4706#ifdef CONFIG_RING_BUFFER_STARTUP_TEST
4707/*
4708 * This is a basic integrity check of the ring buffer.
4709 * Late in the boot cycle this test will run when configured in.
4710 * It will kick off a thread per CPU that will go into a loop
4711 * writing to the per cpu ring buffer various sizes of data.
4712 * Some of the data will be large items, some small.
4713 *
4714 * Another thread is created that goes into a spin, sending out
4715 * IPIs to the other CPUs to also write into the ring buffer.
4716 * this is to test the nesting ability of the buffer.
4717 *
4718 * Basic stats are recorded and reported. If something in the
4719 * ring buffer should happen that's not expected, a big warning
4720 * is displayed and all ring buffers are disabled.
4721 */
4722static struct task_struct *rb_threads[NR_CPUS] __initdata;
4723
4724struct rb_test_data {
4725        struct ring_buffer      *buffer;
4726        unsigned long           events;
4727        unsigned long           bytes_written;
4728        unsigned long           bytes_alloc;
4729        unsigned long           bytes_dropped;
4730        unsigned long           events_nested;
4731        unsigned long           bytes_written_nested;
4732        unsigned long           bytes_alloc_nested;
4733        unsigned long           bytes_dropped_nested;
4734        int                     min_size_nested;
4735        int                     max_size_nested;
4736        int                     max_size;
4737        int                     min_size;
4738        int                     cpu;
4739        int                     cnt;
4740};
4741
4742static struct rb_test_data rb_data[NR_CPUS] __initdata;
4743
4744/* 1 meg per cpu */
4745#define RB_TEST_BUFFER_SIZE     1048576
4746
4747static char rb_string[] __initdata =
4748        "abcdefghijklmnopqrstuvwxyz1234567890!@#$%^&*()?+\\"
4749        "?+|:';\",.<>/?abcdefghijklmnopqrstuvwxyz1234567890"
4750        "!@#$%^&*()?+\\?+|:';\",.<>/?abcdefghijklmnopqrstuv";
4751
4752static bool rb_test_started __initdata;
4753
4754struct rb_item {
4755        int size;
4756        char str[];
4757};
4758
4759static __init int rb_write_something(struct rb_test_data *data, bool nested)
4760{
4761        struct ring_buffer_event *event;
4762        struct rb_item *item;
4763        bool started;
4764        int event_len;
4765        int size;
4766        int len;
4767        int cnt;
4768
4769        /* Have nested writes different that what is written */
4770        cnt = data->cnt + (nested ? 27 : 0);
4771
4772        /* Multiply cnt by ~e, to make some unique increment */
4773        size = (data->cnt * 68 / 25) % (sizeof(rb_string) - 1);
4774
4775        len = size + sizeof(struct rb_item);
4776
4777        started = rb_test_started;
4778        /* read rb_test_started before checking buffer enabled */
4779        smp_rmb();
4780
4781        event = ring_buffer_lock_reserve(data->buffer, len);
4782        if (!event) {
4783                /* Ignore dropped events before test starts. */
4784                if (started) {
4785                        if (nested)
4786                                data->bytes_dropped += len;
4787                        else
4788                                data->bytes_dropped_nested += len;
4789                }
4790                return len;
4791        }
4792
4793        event_len = ring_buffer_event_length(event);
4794
4795        if (RB_WARN_ON(data->buffer, event_len < len))
4796                goto out;
4797
4798        item = ring_buffer_event_data(event);
4799        item->size = size;
4800        memcpy(item->str, rb_string, size);
4801
4802        if (nested) {
4803                data->bytes_alloc_nested += event_len;
4804                data->bytes_written_nested += len;
4805                data->events_nested++;
4806                if (!data->min_size_nested || len < data->min_size_nested)
4807                        data->min_size_nested = len;
4808                if (len > data->max_size_nested)
4809                        data->max_size_nested = len;
4810        } else {
4811                data->bytes_alloc += event_len;
4812                data->bytes_written += len;
4813                data->events++;
4814                if (!data->min_size || len < data->min_size)
4815                        data->max_size = len;
4816                if (len > data->max_size)
4817                        data->max_size = len;
4818        }
4819
4820 out:
4821        ring_buffer_unlock_commit(data->buffer, event);
4822
4823        return 0;
4824}
4825
4826static __init int rb_test(void *arg)
4827{
4828        struct rb_test_data *data = arg;
4829
4830        while (!kthread_should_stop()) {
4831                rb_write_something(data, false);
4832                data->cnt++;
4833
4834                set_current_state(TASK_INTERRUPTIBLE);
4835                /* Now sleep between a min of 100-300us and a max of 1ms */
4836                usleep_range(((data->cnt % 3) + 1) * 100, 1000);
4837        }
4838
4839        return 0;
4840}
4841
4842static __init void rb_ipi(void *ignore)
4843{
4844        struct rb_test_data *data;
4845        int cpu = smp_processor_id();
4846
4847        data = &rb_data[cpu];
4848        rb_write_something(data, true);
4849}
4850
4851static __init int rb_hammer_test(void *arg)
4852{
4853        while (!kthread_should_stop()) {
4854
4855                /* Send an IPI to all cpus to write data! */
4856                smp_call_function(rb_ipi, NULL, 1);
4857                /* No sleep, but for non preempt, let others run */
4858                schedule();
4859        }
4860
4861        return 0;
4862}
4863
4864static __init int test_ringbuffer(void)
4865{
4866        struct task_struct *rb_hammer;
4867        struct ring_buffer *buffer;
4868        int cpu;
4869        int ret = 0;
4870
4871        pr_info("Running ring buffer tests...\n");
4872
4873        buffer = ring_buffer_alloc(RB_TEST_BUFFER_SIZE, RB_FL_OVERWRITE);
4874        if (WARN_ON(!buffer))
4875                return 0;
4876
4877        /* Disable buffer so that threads can't write to it yet */
4878        ring_buffer_record_off(buffer);
4879
4880        for_each_online_cpu(cpu) {
4881                rb_data[cpu].buffer = buffer;
4882                rb_data[cpu].cpu = cpu;
4883                rb_data[cpu].cnt = cpu;
4884                rb_threads[cpu] = kthread_create(rb_test, &rb_data[cpu],
4885                                                 "rbtester/%d", cpu);
4886                if (WARN_ON(IS_ERR(rb_threads[cpu]))) {
4887                        pr_cont("FAILED\n");
4888                        ret = PTR_ERR(rb_threads[cpu]);
4889                        goto out_free;
4890                }
4891
4892                kthread_bind(rb_threads[cpu], cpu);
4893                wake_up_process(rb_threads[cpu]);
4894        }
4895
4896        /* Now create the rb hammer! */
4897        rb_hammer = kthread_run(rb_hammer_test, NULL, "rbhammer");
4898        if (WARN_ON(IS_ERR(rb_hammer))) {
4899                pr_cont("FAILED\n");
4900                ret = PTR_ERR(rb_hammer);
4901                goto out_free;
4902        }
4903
4904        ring_buffer_record_on(buffer);
4905        /*
4906         * Show buffer is enabled before setting rb_test_started.
4907         * Yes there's a small race window where events could be
4908         * dropped and the thread wont catch it. But when a ring
4909         * buffer gets enabled, there will always be some kind of
4910         * delay before other CPUs see it. Thus, we don't care about
4911         * those dropped events. We care about events dropped after
4912         * the threads see that the buffer is active.
4913         */
4914        smp_wmb();
4915        rb_test_started = true;
4916
4917        set_current_state(TASK_INTERRUPTIBLE);
4918        /* Just run for 10 seconds */;
4919        schedule_timeout(10 * HZ);
4920
4921        kthread_stop(rb_hammer);
4922
4923 out_free:
4924        for_each_online_cpu(cpu) {
4925                if (!rb_threads[cpu])
4926                        break;
4927                kthread_stop(rb_threads[cpu]);
4928        }
4929        if (ret) {
4930                ring_buffer_free(buffer);
4931                return ret;
4932        }
4933
4934        /* Report! */
4935        pr_info("finished\n");
4936        for_each_online_cpu(cpu) {
4937                struct ring_buffer_event *event;
4938                struct rb_test_data *data = &rb_data[cpu];
4939                struct rb_item *item;
4940                unsigned long total_events;
4941                unsigned long total_dropped;
4942                unsigned long total_written;
4943                unsigned long total_alloc;
4944                unsigned long total_read = 0;
4945                unsigned long total_size = 0;
4946                unsigned long total_len = 0;
4947                unsigned long total_lost = 0;
4948                unsigned long lost;
4949                int big_event_size;
4950                int small_event_size;
4951
4952                ret = -1;
4953
4954                total_events = data->events + data->events_nested;
4955                total_written = data->bytes_written + data->bytes_written_nested;
4956                total_alloc = data->bytes_alloc + data->bytes_alloc_nested;
4957                total_dropped = data->bytes_dropped + data->bytes_dropped_nested;
4958
4959                big_event_size = data->max_size + data->max_size_nested;
4960                small_event_size = data->min_size + data->min_size_nested;
4961
4962                pr_info("CPU %d:\n", cpu);
4963                pr_info("              events:    %ld\n", total_events);
4964                pr_info("       dropped bytes:    %ld\n", total_dropped);
4965                pr_info("       alloced bytes:    %ld\n", total_alloc);
4966                pr_info("       written bytes:    %ld\n", total_written);
4967                pr_info("       biggest event:    %d\n", big_event_size);
4968                pr_info("      smallest event:    %d\n", small_event_size);
4969
4970                if (RB_WARN_ON(buffer, total_dropped))
4971                        break;
4972
4973                ret = 0;
4974
4975                while ((event = ring_buffer_consume(buffer, cpu, NULL, &lost))) {
4976                        total_lost += lost;
4977                        item = ring_buffer_event_data(event);
4978                        total_len += ring_buffer_event_length(event);
4979                        total_size += item->size + sizeof(struct rb_item);
4980                        if (memcmp(&item->str[0], rb_string, item->size) != 0) {
4981                                pr_info("FAILED!\n");
4982                                pr_info("buffer had: %.*s\n", item->size, item->str);
4983                                pr_info("expected:   %.*s\n", item->size, rb_string);
4984                                RB_WARN_ON(buffer, 1);
4985                                ret = -1;
4986                                break;
4987                        }
4988                        total_read++;
4989                }
4990                if (ret)
4991                        break;
4992
4993                ret = -1;
4994
4995                pr_info("         read events:   %ld\n", total_read);
4996                pr_info("         lost events:   %ld\n", total_lost);
4997                pr_info("        total events:   %ld\n", total_lost + total_read);
4998                pr_info("  recorded len bytes:   %ld\n", total_len);
4999                pr_info(" recorded size bytes:   %ld\n", total_size);
5000                if (total_lost)
5001                        pr_info(" With dropped events, record len and size may not match\n"
5002                                " alloced and written from above\n");
5003                if (!total_lost) {
5004                        if (RB_WARN_ON(buffer, total_len != total_alloc ||
5005                                       total_size != total_written))
5006                                break;
5007                }
5008                if (RB_WARN_ON(buffer, total_lost + total_read != total_events))
5009                        break;
5010
5011                ret = 0;
5012        }
5013        if (!ret)
5014                pr_info("Ring buffer PASSED!\n");
5015
5016        ring_buffer_free(buffer);
5017        return 0;
5018}
5019
5020late_initcall(test_ringbuffer);
5021#endif /* CONFIG_RING_BUFFER_STARTUP_TEST */
Note: See TracBrowser for help on using the repository browser.