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