648f25aad84ca6b0016a08016346b414a517ce0f
[pandora-kernel.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/ring_buffer.h>
7 #include <linux/trace_clock.h>
8 #include <linux/spinlock.h>
9 #include <linux/debugfs.h>
10 #include <linux/uaccess.h>
11 #include <linux/hardirq.h>
12 #include <linux/kmemcheck.h>
13 #include <linux/module.h>
14 #include <linux/percpu.h>
15 #include <linux/mutex.h>
16 #include <linux/slab.h>
17 #include <linux/init.h>
18 #include <linux/hash.h>
19 #include <linux/list.h>
20 #include <linux/cpu.h>
21 #include <linux/fs.h>
22
23 #include <asm/local.h>
24 #include "trace.h"
25
26 /*
27  * The ring buffer header is special. We must manually up keep it.
28  */
29 int ring_buffer_print_entry_header(struct trace_seq *s)
30 {
31         int ret;
32
33         ret = trace_seq_printf(s, "# compressed entry header\n");
34         ret = trace_seq_printf(s, "\ttype_len    :    5 bits\n");
35         ret = trace_seq_printf(s, "\ttime_delta  :   27 bits\n");
36         ret = trace_seq_printf(s, "\tarray       :   32 bits\n");
37         ret = trace_seq_printf(s, "\n");
38         ret = trace_seq_printf(s, "\tpadding     : type == %d\n",
39                                RINGBUF_TYPE_PADDING);
40         ret = trace_seq_printf(s, "\ttime_extend : type == %d\n",
41                                RINGBUF_TYPE_TIME_EXTEND);
42         ret = trace_seq_printf(s, "\tdata max type_len  == %d\n",
43                                RINGBUF_TYPE_DATA_TYPE_LEN_MAX);
44
45         return ret;
46 }
47
48 /*
49  * The ring buffer is made up of a list of pages. A separate list of pages is
50  * allocated for each CPU. A writer may only write to a buffer that is
51  * associated with the CPU it is currently executing on.  A reader may read
52  * from any per cpu buffer.
53  *
54  * The reader is special. For each per cpu buffer, the reader has its own
55  * reader page. When a reader has read the entire reader page, this reader
56  * page is swapped with another page in the ring buffer.
57  *
58  * Now, as long as the writer is off the reader page, the reader can do what
59  * ever it wants with that page. The writer will never write to that page
60  * again (as long as it is out of the ring buffer).
61  *
62  * Here's some silly ASCII art.
63  *
64  *   +------+
65  *   |reader|          RING BUFFER
66  *   |page  |
67  *   +------+        +---+   +---+   +---+
68  *                   |   |-->|   |-->|   |
69  *                   +---+   +---+   +---+
70  *                     ^               |
71  *                     |               |
72  *                     +---------------+
73  *
74  *
75  *   +------+
76  *   |reader|          RING BUFFER
77  *   |page  |------------------v
78  *   +------+        +---+   +---+   +---+
79  *                   |   |-->|   |-->|   |
80  *                   +---+   +---+   +---+
81  *                     ^               |
82  *                     |               |
83  *                     +---------------+
84  *
85  *
86  *   +------+
87  *   |reader|          RING BUFFER
88  *   |page  |------------------v
89  *   +------+        +---+   +---+   +---+
90  *      ^            |   |-->|   |-->|   |
91  *      |            +---+   +---+   +---+
92  *      |                              |
93  *      |                              |
94  *      +------------------------------+
95  *
96  *
97  *   +------+
98  *   |buffer|          RING BUFFER
99  *   |page  |------------------v
100  *   +------+        +---+   +---+   +---+
101  *      ^            |   |   |   |-->|   |
102  *      |   New      +---+   +---+   +---+
103  *      |  Reader------^               |
104  *      |   page                       |
105  *      +------------------------------+
106  *
107  *
108  * After we make this swap, the reader can hand this page off to the splice
109  * code and be done with it. It can even allocate a new page if it needs to
110  * and swap that into the ring buffer.
111  *
112  * We will be using cmpxchg soon to make all this lockless.
113  *
114  */
115
116 /*
117  * A fast way to enable or disable all ring buffers is to
118  * call tracing_on or tracing_off. Turning off the ring buffers
119  * prevents all ring buffers from being recorded to.
120  * Turning this switch on, makes it OK to write to the
121  * ring buffer, if the ring buffer is enabled itself.
122  *
123  * There's three layers that must be on in order to write
124  * to the ring buffer.
125  *
126  * 1) This global flag must be set.
127  * 2) The ring buffer must be enabled for recording.
128  * 3) The per cpu buffer must be enabled for recording.
129  *
130  * In case of an anomaly, this global flag has a bit set that
131  * will permantly disable all ring buffers.
132  */
133
134 /*
135  * Global flag to disable all recording to ring buffers
136  *  This has two bits: ON, DISABLED
137  *
138  *  ON   DISABLED
139  * ---- ----------
140  *   0      0        : ring buffers are off
141  *   1      0        : ring buffers are on
142  *   X      1        : ring buffers are permanently disabled
143  */
144
145 enum {
146         RB_BUFFERS_ON_BIT       = 0,
147         RB_BUFFERS_DISABLED_BIT = 1,
148 };
149
150 enum {
151         RB_BUFFERS_ON           = 1 << RB_BUFFERS_ON_BIT,
152         RB_BUFFERS_DISABLED     = 1 << RB_BUFFERS_DISABLED_BIT,
153 };
154
155 static unsigned long ring_buffer_flags __read_mostly = RB_BUFFERS_ON;
156
157 #define BUF_PAGE_HDR_SIZE offsetof(struct buffer_data_page, data)
158
159 /**
160  * tracing_on - enable all tracing buffers
161  *
162  * This function enables all tracing buffers that may have been
163  * disabled with tracing_off.
164  */
165 void tracing_on(void)
166 {
167         set_bit(RB_BUFFERS_ON_BIT, &ring_buffer_flags);
168 }
169 EXPORT_SYMBOL_GPL(tracing_on);
170
171 /**
172  * tracing_off - turn off all tracing buffers
173  *
174  * This function stops all tracing buffers from recording data.
175  * It does not disable any overhead the tracers themselves may
176  * be causing. This function simply causes all recording to
177  * the ring buffers to fail.
178  */
179 void tracing_off(void)
180 {
181         clear_bit(RB_BUFFERS_ON_BIT, &ring_buffer_flags);
182 }
183 EXPORT_SYMBOL_GPL(tracing_off);
184
185 /**
186  * tracing_off_permanent - permanently disable ring buffers
187  *
188  * This function, once called, will disable all ring buffers
189  * permanently.
190  */
191 void tracing_off_permanent(void)
192 {
193         set_bit(RB_BUFFERS_DISABLED_BIT, &ring_buffer_flags);
194 }
195
196 /**
197  * tracing_is_on - show state of ring buffers enabled
198  */
199 int tracing_is_on(void)
200 {
201         return ring_buffer_flags == RB_BUFFERS_ON;
202 }
203 EXPORT_SYMBOL_GPL(tracing_is_on);
204
205 #define RB_EVNT_HDR_SIZE (offsetof(struct ring_buffer_event, array))
206 #define RB_ALIGNMENT            4U
207 #define RB_MAX_SMALL_DATA       (RB_ALIGNMENT * RINGBUF_TYPE_DATA_TYPE_LEN_MAX)
208 #define RB_EVNT_MIN_SIZE        8U      /* two 32bit words */
209
210 #if !defined(CONFIG_64BIT) || defined(CONFIG_HAVE_EFFICIENT_UNALIGNED_ACCESS)
211 # define RB_FORCE_8BYTE_ALIGNMENT       0
212 # define RB_ARCH_ALIGNMENT              RB_ALIGNMENT
213 #else
214 # define RB_FORCE_8BYTE_ALIGNMENT       1
215 # define RB_ARCH_ALIGNMENT              8U
216 #endif
217
218 /* define RINGBUF_TYPE_DATA for 'case RINGBUF_TYPE_DATA:' */
219 #define RINGBUF_TYPE_DATA 0 ... RINGBUF_TYPE_DATA_TYPE_LEN_MAX
220
221 enum {
222         RB_LEN_TIME_EXTEND = 8,
223         RB_LEN_TIME_STAMP = 16,
224 };
225
226 #define skip_time_extend(event) \
227         ((struct ring_buffer_event *)((char *)event + RB_LEN_TIME_EXTEND))
228
229 static inline int rb_null_event(struct ring_buffer_event *event)
230 {
231         return event->type_len == RINGBUF_TYPE_PADDING && !event->time_delta;
232 }
233
234 static void rb_event_set_padding(struct ring_buffer_event *event)
235 {
236         /* padding has a NULL time_delta */
237         event->type_len = RINGBUF_TYPE_PADDING;
238         event->time_delta = 0;
239 }
240
241 static unsigned
242 rb_event_data_length(struct ring_buffer_event *event)
243 {
244         unsigned length;
245
246         if (event->type_len)
247                 length = event->type_len * RB_ALIGNMENT;
248         else
249                 length = event->array[0];
250         return length + RB_EVNT_HDR_SIZE;
251 }
252
253 /*
254  * Return the length of the given event. Will return
255  * the length of the time extend if the event is a
256  * time extend.
257  */
258 static inline unsigned
259 rb_event_length(struct ring_buffer_event *event)
260 {
261         switch (event->type_len) {
262         case RINGBUF_TYPE_PADDING:
263                 if (rb_null_event(event))
264                         /* undefined */
265                         return -1;
266                 return  event->array[0] + RB_EVNT_HDR_SIZE;
267
268         case RINGBUF_TYPE_TIME_EXTEND:
269                 return RB_LEN_TIME_EXTEND;
270
271         case RINGBUF_TYPE_TIME_STAMP:
272                 return RB_LEN_TIME_STAMP;
273
274         case RINGBUF_TYPE_DATA:
275                 return rb_event_data_length(event);
276         default:
277                 BUG();
278         }
279         /* not hit */
280         return 0;
281 }
282
283 /*
284  * Return total length of time extend and data,
285  *   or just the event length for all other events.
286  */
287 static inline unsigned
288 rb_event_ts_length(struct ring_buffer_event *event)
289 {
290         unsigned len = 0;
291
292         if (event->type_len == RINGBUF_TYPE_TIME_EXTEND) {
293                 /* time extends include the data event after it */
294                 len = RB_LEN_TIME_EXTEND;
295                 event = skip_time_extend(event);
296         }
297         return len + rb_event_length(event);
298 }
299
300 /**
301  * ring_buffer_event_length - return the length of the event
302  * @event: the event to get the length of
303  *
304  * Returns the size of the data load of a data event.
305  * If the event is something other than a data event, it
306  * returns the size of the event itself. With the exception
307  * of a TIME EXTEND, where it still returns the size of the
308  * data load of the data event after it.
309  */
310 unsigned ring_buffer_event_length(struct ring_buffer_event *event)
311 {
312         unsigned length;
313
314         if (event->type_len == RINGBUF_TYPE_TIME_EXTEND)
315                 event = skip_time_extend(event);
316
317         length = rb_event_length(event);
318         if (event->type_len > RINGBUF_TYPE_DATA_TYPE_LEN_MAX)
319                 return length;
320         length -= RB_EVNT_HDR_SIZE;
321         if (length > RB_MAX_SMALL_DATA + sizeof(event->array[0]))
322                 length -= sizeof(event->array[0]);
323         return length;
324 }
325 EXPORT_SYMBOL_GPL(ring_buffer_event_length);
326
327 /* inline for ring buffer fast paths */
328 static void *
329 rb_event_data(struct ring_buffer_event *event)
330 {
331         if (event->type_len == RINGBUF_TYPE_TIME_EXTEND)
332                 event = skip_time_extend(event);
333         BUG_ON(event->type_len > RINGBUF_TYPE_DATA_TYPE_LEN_MAX);
334         /* If length is in len field, then array[0] has the data */
335         if (event->type_len)
336                 return (void *)&event->array[0];
337         /* Otherwise length is in array[0] and array[1] has the data */
338         return (void *)&event->array[1];
339 }
340
341 /**
342  * ring_buffer_event_data - return the data of the event
343  * @event: the event to get the data from
344  */
345 void *ring_buffer_event_data(struct ring_buffer_event *event)
346 {
347         return rb_event_data(event);
348 }
349 EXPORT_SYMBOL_GPL(ring_buffer_event_data);
350
351 #define for_each_buffer_cpu(buffer, cpu)                \
352         for_each_cpu(cpu, buffer->cpumask)
353
354 #define TS_SHIFT        27
355 #define TS_MASK         ((1ULL << TS_SHIFT) - 1)
356 #define TS_DELTA_TEST   (~TS_MASK)
357
358 /* Flag when events were overwritten */
359 #define RB_MISSED_EVENTS        (1 << 31)
360 /* Missed count stored at end */
361 #define RB_MISSED_STORED        (1 << 30)
362
363 struct buffer_data_page {
364         u64              time_stamp;    /* page time stamp */
365         local_t          commit;        /* write committed index */
366         unsigned char    data[];        /* data of buffer page */
367 };
368
369 /*
370  * Note, the buffer_page list must be first. The buffer pages
371  * are allocated in cache lines, which means that each buffer
372  * page will be at the beginning of a cache line, and thus
373  * the least significant bits will be zero. We use this to
374  * add flags in the list struct pointers, to make the ring buffer
375  * lockless.
376  */
377 struct buffer_page {
378         struct list_head list;          /* list of buffer pages */
379         local_t          write;         /* index for next write */
380         unsigned         read;          /* index for next read */
381         local_t          entries;       /* entries on this page */
382         unsigned long    real_end;      /* real end of data */
383         struct buffer_data_page *page;  /* Actual data page */
384 };
385
386 /*
387  * The buffer page counters, write and entries, must be reset
388  * atomically when crossing page boundaries. To synchronize this
389  * update, two counters are inserted into the number. One is
390  * the actual counter for the write position or count on the page.
391  *
392  * The other is a counter of updaters. Before an update happens
393  * the update partition of the counter is incremented. This will
394  * allow the updater to update the counter atomically.
395  *
396  * The counter is 20 bits, and the state data is 12.
397  */
398 #define RB_WRITE_MASK           0xfffff
399 #define RB_WRITE_INTCNT         (1 << 20)
400
401 static void rb_init_page(struct buffer_data_page *bpage)
402 {
403         local_set(&bpage->commit, 0);
404 }
405
406 /**
407  * ring_buffer_page_len - the size of data on the page.
408  * @page: The page to read
409  *
410  * Returns the amount of data on the page, including buffer page header.
411  */
412 size_t ring_buffer_page_len(void *page)
413 {
414         return local_read(&((struct buffer_data_page *)page)->commit)
415                 + BUF_PAGE_HDR_SIZE;
416 }
417
418 /*
419  * Also stolen from mm/slob.c. Thanks to Mathieu Desnoyers for pointing
420  * this issue out.
421  */
422 static void free_buffer_page(struct buffer_page *bpage)
423 {
424         free_page((unsigned long)bpage->page);
425         kfree(bpage);
426 }
427
428 /*
429  * We need to fit the time_stamp delta into 27 bits.
430  */
431 static inline int test_time_stamp(u64 delta)
432 {
433         if (delta & TS_DELTA_TEST)
434                 return 1;
435         return 0;
436 }
437
438 #define BUF_PAGE_SIZE (PAGE_SIZE - BUF_PAGE_HDR_SIZE)
439
440 /* Max payload is BUF_PAGE_SIZE - header (8bytes) */
441 #define BUF_MAX_DATA_SIZE (BUF_PAGE_SIZE - (sizeof(u32) * 2))
442
443 int ring_buffer_print_page_header(struct trace_seq *s)
444 {
445         struct buffer_data_page field;
446         int ret;
447
448         ret = trace_seq_printf(s, "\tfield: u64 timestamp;\t"
449                                "offset:0;\tsize:%u;\tsigned:%u;\n",
450                                (unsigned int)sizeof(field.time_stamp),
451                                (unsigned int)is_signed_type(u64));
452
453         ret = trace_seq_printf(s, "\tfield: local_t commit;\t"
454                                "offset:%u;\tsize:%u;\tsigned:%u;\n",
455                                (unsigned int)offsetof(typeof(field), commit),
456                                (unsigned int)sizeof(field.commit),
457                                (unsigned int)is_signed_type(long));
458
459         ret = trace_seq_printf(s, "\tfield: int overwrite;\t"
460                                "offset:%u;\tsize:%u;\tsigned:%u;\n",
461                                (unsigned int)offsetof(typeof(field), commit),
462                                1,
463                                (unsigned int)is_signed_type(long));
464
465         ret = trace_seq_printf(s, "\tfield: char data;\t"
466                                "offset:%u;\tsize:%u;\tsigned:%u;\n",
467                                (unsigned int)offsetof(typeof(field), data),
468                                (unsigned int)BUF_PAGE_SIZE,
469                                (unsigned int)is_signed_type(char));
470
471         return ret;
472 }
473
474 /*
475  * head_page == tail_page && head == tail then buffer is empty.
476  */
477 struct ring_buffer_per_cpu {
478         int                             cpu;
479         atomic_t                        record_disabled;
480         struct ring_buffer              *buffer;
481         raw_spinlock_t                  reader_lock;    /* serialize readers */
482         arch_spinlock_t                 lock;
483         struct lock_class_key           lock_key;
484         struct list_head                *pages;
485         struct buffer_page              *head_page;     /* read from head */
486         struct buffer_page              *tail_page;     /* write to tail */
487         struct buffer_page              *commit_page;   /* committed pages */
488         struct buffer_page              *reader_page;
489         unsigned long                   lost_events;
490         unsigned long                   last_overrun;
491         local_t                         entries_bytes;
492         local_t                         commit_overrun;
493         local_t                         overrun;
494         local_t                         entries;
495         local_t                         committing;
496         local_t                         commits;
497         unsigned long                   read;
498         unsigned long                   read_bytes;
499         u64                             write_stamp;
500         u64                             read_stamp;
501 };
502
503 struct ring_buffer {
504         unsigned                        pages;
505         unsigned                        flags;
506         int                             cpus;
507         atomic_t                        record_disabled;
508         cpumask_var_t                   cpumask;
509
510         struct lock_class_key           *reader_lock_key;
511
512         struct mutex                    mutex;
513
514         struct ring_buffer_per_cpu      **buffers;
515
516 #ifdef CONFIG_HOTPLUG_CPU
517         struct notifier_block           cpu_notify;
518 #endif
519         u64                             (*clock)(void);
520 };
521
522 struct ring_buffer_iter {
523         struct ring_buffer_per_cpu      *cpu_buffer;
524         unsigned long                   head;
525         struct buffer_page              *head_page;
526         struct buffer_page              *cache_reader_page;
527         unsigned long                   cache_read;
528         u64                             read_stamp;
529 };
530
531 /* buffer may be either ring_buffer or ring_buffer_per_cpu */
532 #define RB_WARN_ON(b, cond)                                             \
533         ({                                                              \
534                 int _____ret = unlikely(cond);                          \
535                 if (_____ret) {                                         \
536                         if (__same_type(*(b), struct ring_buffer_per_cpu)) { \
537                                 struct ring_buffer_per_cpu *__b =       \
538                                         (void *)b;                      \
539                                 atomic_inc(&__b->buffer->record_disabled); \
540                         } else                                          \
541                                 atomic_inc(&b->record_disabled);        \
542                         WARN_ON(1);                                     \
543                 }                                                       \
544                 _____ret;                                               \
545         })
546
547 /* Up this if you want to test the TIME_EXTENTS and normalization */
548 #define DEBUG_SHIFT 0
549
550 static inline u64 rb_time_stamp(struct ring_buffer *buffer)
551 {
552         /* shift to debug/test normalization and TIME_EXTENTS */
553         return buffer->clock() << DEBUG_SHIFT;
554 }
555
556 u64 ring_buffer_time_stamp(struct ring_buffer *buffer, int cpu)
557 {
558         u64 time;
559
560         preempt_disable_notrace();
561         time = rb_time_stamp(buffer);
562         preempt_enable_no_resched_notrace();
563
564         return time;
565 }
566 EXPORT_SYMBOL_GPL(ring_buffer_time_stamp);
567
568 void ring_buffer_normalize_time_stamp(struct ring_buffer *buffer,
569                                       int cpu, u64 *ts)
570 {
571         /* Just stupid testing the normalize function and deltas */
572         *ts >>= DEBUG_SHIFT;
573 }
574 EXPORT_SYMBOL_GPL(ring_buffer_normalize_time_stamp);
575
576 /*
577  * Making the ring buffer lockless makes things tricky.
578  * Although writes only happen on the CPU that they are on,
579  * and they only need to worry about interrupts. Reads can
580  * happen on any CPU.
581  *
582  * The reader page is always off the ring buffer, but when the
583  * reader finishes with a page, it needs to swap its page with
584  * a new one from the buffer. The reader needs to take from
585  * the head (writes go to the tail). But if a writer is in overwrite
586  * mode and wraps, it must push the head page forward.
587  *
588  * Here lies the problem.
589  *
590  * The reader must be careful to replace only the head page, and
591  * not another one. As described at the top of the file in the
592  * ASCII art, the reader sets its old page to point to the next
593  * page after head. It then sets the page after head to point to
594  * the old reader page. But if the writer moves the head page
595  * during this operation, the reader could end up with the tail.
596  *
597  * We use cmpxchg to help prevent this race. We also do something
598  * special with the page before head. We set the LSB to 1.
599  *
600  * When the writer must push the page forward, it will clear the
601  * bit that points to the head page, move the head, and then set
602  * the bit that points to the new head page.
603  *
604  * We also don't want an interrupt coming in and moving the head
605  * page on another writer. Thus we use the second LSB to catch
606  * that too. Thus:
607  *
608  * head->list->prev->next        bit 1          bit 0
609  *                              -------        -------
610  * Normal page                     0              0
611  * Points to head page             0              1
612  * New head page                   1              0
613  *
614  * Note we can not trust the prev pointer of the head page, because:
615  *
616  * +----+       +-----+        +-----+
617  * |    |------>|  T  |---X--->|  N  |
618  * |    |<------|     |        |     |
619  * +----+       +-----+        +-----+
620  *   ^                           ^ |
621  *   |          +-----+          | |
622  *   +----------|  R  |----------+ |
623  *              |     |<-----------+
624  *              +-----+
625  *
626  * Key:  ---X-->  HEAD flag set in pointer
627  *         T      Tail page
628  *         R      Reader page
629  *         N      Next page
630  *
631  * (see __rb_reserve_next() to see where this happens)
632  *
633  *  What the above shows is that the reader just swapped out
634  *  the reader page with a page in the buffer, but before it
635  *  could make the new header point back to the new page added
636  *  it was preempted by a writer. The writer moved forward onto
637  *  the new page added by the reader and is about to move forward
638  *  again.
639  *
640  *  You can see, it is legitimate for the previous pointer of
641  *  the head (or any page) not to point back to itself. But only
642  *  temporarially.
643  */
644
645 #define RB_PAGE_NORMAL          0UL
646 #define RB_PAGE_HEAD            1UL
647 #define RB_PAGE_UPDATE          2UL
648
649
650 #define RB_FLAG_MASK            3UL
651
652 /* PAGE_MOVED is not part of the mask */
653 #define RB_PAGE_MOVED           4UL
654
655 /*
656  * rb_list_head - remove any bit
657  */
658 static struct list_head *rb_list_head(struct list_head *list)
659 {
660         unsigned long val = (unsigned long)list;
661
662         return (struct list_head *)(val & ~RB_FLAG_MASK);
663 }
664
665 /*
666  * rb_is_head_page - test if the given page is the head page
667  *
668  * Because the reader may move the head_page pointer, we can
669  * not trust what the head page is (it may be pointing to
670  * the reader page). But if the next page is a header page,
671  * its flags will be non zero.
672  */
673 static inline int
674 rb_is_head_page(struct ring_buffer_per_cpu *cpu_buffer,
675                 struct buffer_page *page, struct list_head *list)
676 {
677         unsigned long val;
678
679         val = (unsigned long)list->next;
680
681         if ((val & ~RB_FLAG_MASK) != (unsigned long)&page->list)
682                 return RB_PAGE_MOVED;
683
684         return val & RB_FLAG_MASK;
685 }
686
687 /*
688  * rb_is_reader_page
689  *
690  * The unique thing about the reader page, is that, if the
691  * writer is ever on it, the previous pointer never points
692  * back to the reader page.
693  */
694 static int rb_is_reader_page(struct buffer_page *page)
695 {
696         struct list_head *list = page->list.prev;
697
698         return rb_list_head(list->next) != &page->list;
699 }
700
701 /*
702  * rb_set_list_to_head - set a list_head to be pointing to head.
703  */
704 static void rb_set_list_to_head(struct ring_buffer_per_cpu *cpu_buffer,
705                                 struct list_head *list)
706 {
707         unsigned long *ptr;
708
709         ptr = (unsigned long *)&list->next;
710         *ptr |= RB_PAGE_HEAD;
711         *ptr &= ~RB_PAGE_UPDATE;
712 }
713
714 /*
715  * rb_head_page_activate - sets up head page
716  */
717 static void rb_head_page_activate(struct ring_buffer_per_cpu *cpu_buffer)
718 {
719         struct buffer_page *head;
720
721         head = cpu_buffer->head_page;
722         if (!head)
723                 return;
724
725         /*
726          * Set the previous list pointer to have the HEAD flag.
727          */
728         rb_set_list_to_head(cpu_buffer, head->list.prev);
729 }
730
731 static void rb_list_head_clear(struct list_head *list)
732 {
733         unsigned long *ptr = (unsigned long *)&list->next;
734
735         *ptr &= ~RB_FLAG_MASK;
736 }
737
738 /*
739  * rb_head_page_dactivate - clears head page ptr (for free list)
740  */
741 static void
742 rb_head_page_deactivate(struct ring_buffer_per_cpu *cpu_buffer)
743 {
744         struct list_head *hd;
745
746         /* Go through the whole list and clear any pointers found. */
747         rb_list_head_clear(cpu_buffer->pages);
748
749         list_for_each(hd, cpu_buffer->pages)
750                 rb_list_head_clear(hd);
751 }
752
753 static int rb_head_page_set(struct ring_buffer_per_cpu *cpu_buffer,
754                             struct buffer_page *head,
755                             struct buffer_page *prev,
756                             int old_flag, int new_flag)
757 {
758         struct list_head *list;
759         unsigned long val = (unsigned long)&head->list;
760         unsigned long ret;
761
762         list = &prev->list;
763
764         val &= ~RB_FLAG_MASK;
765
766         ret = cmpxchg((unsigned long *)&list->next,
767                       val | old_flag, val | new_flag);
768
769         /* check if the reader took the page */
770         if ((ret & ~RB_FLAG_MASK) != val)
771                 return RB_PAGE_MOVED;
772
773         return ret & RB_FLAG_MASK;
774 }
775
776 static int rb_head_page_set_update(struct ring_buffer_per_cpu *cpu_buffer,
777                                    struct buffer_page *head,
778                                    struct buffer_page *prev,
779                                    int old_flag)
780 {
781         return rb_head_page_set(cpu_buffer, head, prev,
782                                 old_flag, RB_PAGE_UPDATE);
783 }
784
785 static int rb_head_page_set_head(struct ring_buffer_per_cpu *cpu_buffer,
786                                  struct buffer_page *head,
787                                  struct buffer_page *prev,
788                                  int old_flag)
789 {
790         return rb_head_page_set(cpu_buffer, head, prev,
791                                 old_flag, RB_PAGE_HEAD);
792 }
793
794 static int rb_head_page_set_normal(struct ring_buffer_per_cpu *cpu_buffer,
795                                    struct buffer_page *head,
796                                    struct buffer_page *prev,
797                                    int old_flag)
798 {
799         return rb_head_page_set(cpu_buffer, head, prev,
800                                 old_flag, RB_PAGE_NORMAL);
801 }
802
803 static inline void rb_inc_page(struct ring_buffer_per_cpu *cpu_buffer,
804                                struct buffer_page **bpage)
805 {
806         struct list_head *p = rb_list_head((*bpage)->list.next);
807
808         *bpage = list_entry(p, struct buffer_page, list);
809 }
810
811 static struct buffer_page *
812 rb_set_head_page(struct ring_buffer_per_cpu *cpu_buffer)
813 {
814         struct buffer_page *head;
815         struct buffer_page *page;
816         struct list_head *list;
817         int i;
818
819         if (RB_WARN_ON(cpu_buffer, !cpu_buffer->head_page))
820                 return NULL;
821
822         /* sanity check */
823         list = cpu_buffer->pages;
824         if (RB_WARN_ON(cpu_buffer, rb_list_head(list->prev->next) != list))
825                 return NULL;
826
827         page = head = cpu_buffer->head_page;
828         /*
829          * It is possible that the writer moves the header behind
830          * where we started, and we miss in one loop.
831          * A second loop should grab the header, but we'll do
832          * three loops just because I'm paranoid.
833          */
834         for (i = 0; i < 3; i++) {
835                 do {
836                         if (rb_is_head_page(cpu_buffer, page, page->list.prev)) {
837                                 cpu_buffer->head_page = page;
838                                 return page;
839                         }
840                         rb_inc_page(cpu_buffer, &page);
841                 } while (page != head);
842         }
843
844         RB_WARN_ON(cpu_buffer, 1);
845
846         return NULL;
847 }
848
849 static int rb_head_page_replace(struct buffer_page *old,
850                                 struct buffer_page *new)
851 {
852         unsigned long *ptr = (unsigned long *)&old->list.prev->next;
853         unsigned long val;
854         unsigned long ret;
855
856         val = *ptr & ~RB_FLAG_MASK;
857         val |= RB_PAGE_HEAD;
858
859         ret = cmpxchg(ptr, val, (unsigned long)&new->list);
860
861         return ret == val;
862 }
863
864 /*
865  * rb_tail_page_update - move the tail page forward
866  *
867  * Returns 1 if moved tail page, 0 if someone else did.
868  */
869 static int rb_tail_page_update(struct ring_buffer_per_cpu *cpu_buffer,
870                                struct buffer_page *tail_page,
871                                struct buffer_page *next_page)
872 {
873         struct buffer_page *old_tail;
874         unsigned long old_entries;
875         unsigned long old_write;
876         int ret = 0;
877
878         /*
879          * The tail page now needs to be moved forward.
880          *
881          * We need to reset the tail page, but without messing
882          * with possible erasing of data brought in by interrupts
883          * that have moved the tail page and are currently on it.
884          *
885          * We add a counter to the write field to denote this.
886          */
887         old_write = local_add_return(RB_WRITE_INTCNT, &next_page->write);
888         old_entries = local_add_return(RB_WRITE_INTCNT, &next_page->entries);
889
890         /*
891          * Just make sure we have seen our old_write and synchronize
892          * with any interrupts that come in.
893          */
894         barrier();
895
896         /*
897          * If the tail page is still the same as what we think
898          * it is, then it is up to us to update the tail
899          * pointer.
900          */
901         if (tail_page == cpu_buffer->tail_page) {
902                 /* Zero the write counter */
903                 unsigned long val = old_write & ~RB_WRITE_MASK;
904                 unsigned long eval = old_entries & ~RB_WRITE_MASK;
905
906                 /*
907                  * This will only succeed if an interrupt did
908                  * not come in and change it. In which case, we
909                  * do not want to modify it.
910                  *
911                  * We add (void) to let the compiler know that we do not care
912                  * about the return value of these functions. We use the
913                  * cmpxchg to only update if an interrupt did not already
914                  * do it for us. If the cmpxchg fails, we don't care.
915                  */
916                 (void)local_cmpxchg(&next_page->write, old_write, val);
917                 (void)local_cmpxchg(&next_page->entries, old_entries, eval);
918
919                 /*
920                  * No need to worry about races with clearing out the commit.
921                  * it only can increment when a commit takes place. But that
922                  * only happens in the outer most nested commit.
923                  */
924                 local_set(&next_page->page->commit, 0);
925
926                 old_tail = cmpxchg(&cpu_buffer->tail_page,
927                                    tail_page, next_page);
928
929                 if (old_tail == tail_page)
930                         ret = 1;
931         }
932
933         return ret;
934 }
935
936 static int rb_check_bpage(struct ring_buffer_per_cpu *cpu_buffer,
937                           struct buffer_page *bpage)
938 {
939         unsigned long val = (unsigned long)bpage;
940
941         if (RB_WARN_ON(cpu_buffer, val & RB_FLAG_MASK))
942                 return 1;
943
944         return 0;
945 }
946
947 /**
948  * rb_check_list - make sure a pointer to a list has the last bits zero
949  */
950 static int rb_check_list(struct ring_buffer_per_cpu *cpu_buffer,
951                          struct list_head *list)
952 {
953         if (RB_WARN_ON(cpu_buffer, rb_list_head(list->prev) != list->prev))
954                 return 1;
955         if (RB_WARN_ON(cpu_buffer, rb_list_head(list->next) != list->next))
956                 return 1;
957         return 0;
958 }
959
960 /**
961  * check_pages - integrity check of buffer pages
962  * @cpu_buffer: CPU buffer with pages to test
963  *
964  * As a safety measure we check to make sure the data pages have not
965  * been corrupted.
966  */
967 static int rb_check_pages(struct ring_buffer_per_cpu *cpu_buffer)
968 {
969         struct list_head *head = cpu_buffer->pages;
970         struct buffer_page *bpage, *tmp;
971
972         rb_head_page_deactivate(cpu_buffer);
973
974         if (RB_WARN_ON(cpu_buffer, head->next->prev != head))
975                 return -1;
976         if (RB_WARN_ON(cpu_buffer, head->prev->next != head))
977                 return -1;
978
979         if (rb_check_list(cpu_buffer, head))
980                 return -1;
981
982         list_for_each_entry_safe(bpage, tmp, head, list) {
983                 if (RB_WARN_ON(cpu_buffer,
984                                bpage->list.next->prev != &bpage->list))
985                         return -1;
986                 if (RB_WARN_ON(cpu_buffer,
987                                bpage->list.prev->next != &bpage->list))
988                         return -1;
989                 if (rb_check_list(cpu_buffer, &bpage->list))
990                         return -1;
991         }
992
993         rb_head_page_activate(cpu_buffer);
994
995         return 0;
996 }
997
998 static int rb_allocate_pages(struct ring_buffer_per_cpu *cpu_buffer,
999                              unsigned nr_pages)
1000 {
1001         struct buffer_page *bpage, *tmp;
1002         LIST_HEAD(pages);
1003         unsigned i;
1004
1005         WARN_ON(!nr_pages);
1006
1007         for (i = 0; i < nr_pages; i++) {
1008                 struct page *page;
1009                 /*
1010                  * __GFP_NORETRY flag makes sure that the allocation fails
1011                  * gracefully without invoking oom-killer and the system is
1012                  * not destabilized.
1013                  */
1014                 bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),
1015                                     GFP_KERNEL | __GFP_NORETRY,
1016                                     cpu_to_node(cpu_buffer->cpu));
1017                 if (!bpage)
1018                         goto free_pages;
1019
1020                 rb_check_bpage(cpu_buffer, bpage);
1021
1022                 list_add(&bpage->list, &pages);
1023
1024                 page = alloc_pages_node(cpu_to_node(cpu_buffer->cpu),
1025                                         GFP_KERNEL | __GFP_NORETRY, 0);
1026                 if (!page)
1027                         goto free_pages;
1028                 bpage->page = page_address(page);
1029                 rb_init_page(bpage->page);
1030         }
1031
1032         /*
1033          * The ring buffer page list is a circular list that does not
1034          * start and end with a list head. All page list items point to
1035          * other pages.
1036          */
1037         cpu_buffer->pages = pages.next;
1038         list_del(&pages);
1039
1040         rb_check_pages(cpu_buffer);
1041
1042         return 0;
1043
1044  free_pages:
1045         list_for_each_entry_safe(bpage, tmp, &pages, list) {
1046                 list_del_init(&bpage->list);
1047                 free_buffer_page(bpage);
1048         }
1049         return -ENOMEM;
1050 }
1051
1052 static struct ring_buffer_per_cpu *
1053 rb_allocate_cpu_buffer(struct ring_buffer *buffer, int cpu)
1054 {
1055         struct ring_buffer_per_cpu *cpu_buffer;
1056         struct buffer_page *bpage;
1057         struct page *page;
1058         int ret;
1059
1060         cpu_buffer = kzalloc_node(ALIGN(sizeof(*cpu_buffer), cache_line_size()),
1061                                   GFP_KERNEL, cpu_to_node(cpu));
1062         if (!cpu_buffer)
1063                 return NULL;
1064
1065         cpu_buffer->cpu = cpu;
1066         cpu_buffer->buffer = buffer;
1067         raw_spin_lock_init(&cpu_buffer->reader_lock);
1068         lockdep_set_class(&cpu_buffer->reader_lock, buffer->reader_lock_key);
1069         cpu_buffer->lock = (arch_spinlock_t)__ARCH_SPIN_LOCK_UNLOCKED;
1070
1071         bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),
1072                             GFP_KERNEL, cpu_to_node(cpu));
1073         if (!bpage)
1074                 goto fail_free_buffer;
1075
1076         rb_check_bpage(cpu_buffer, bpage);
1077
1078         cpu_buffer->reader_page = bpage;
1079         page = alloc_pages_node(cpu_to_node(cpu), GFP_KERNEL, 0);
1080         if (!page)
1081                 goto fail_free_reader;
1082         bpage->page = page_address(page);
1083         rb_init_page(bpage->page);
1084
1085         INIT_LIST_HEAD(&cpu_buffer->reader_page->list);
1086
1087         ret = rb_allocate_pages(cpu_buffer, buffer->pages);
1088         if (ret < 0)
1089                 goto fail_free_reader;
1090
1091         cpu_buffer->head_page
1092                 = list_entry(cpu_buffer->pages, struct buffer_page, list);
1093         cpu_buffer->tail_page = cpu_buffer->commit_page = cpu_buffer->head_page;
1094
1095         rb_head_page_activate(cpu_buffer);
1096
1097         return cpu_buffer;
1098
1099  fail_free_reader:
1100         free_buffer_page(cpu_buffer->reader_page);
1101
1102  fail_free_buffer:
1103         kfree(cpu_buffer);
1104         return NULL;
1105 }
1106
1107 static void rb_free_cpu_buffer(struct ring_buffer_per_cpu *cpu_buffer)
1108 {
1109         struct list_head *head = cpu_buffer->pages;
1110         struct buffer_page *bpage, *tmp;
1111
1112         free_buffer_page(cpu_buffer->reader_page);
1113
1114         rb_head_page_deactivate(cpu_buffer);
1115
1116         if (head) {
1117                 list_for_each_entry_safe(bpage, tmp, head, list) {
1118                         list_del_init(&bpage->list);
1119                         free_buffer_page(bpage);
1120                 }
1121                 bpage = list_entry(head, struct buffer_page, list);
1122                 free_buffer_page(bpage);
1123         }
1124
1125         kfree(cpu_buffer);
1126 }
1127
1128 #ifdef CONFIG_HOTPLUG_CPU
1129 static int rb_cpu_notify(struct notifier_block *self,
1130                          unsigned long action, void *hcpu);
1131 #endif
1132
1133 /**
1134  * ring_buffer_alloc - allocate a new ring_buffer
1135  * @size: the size in bytes per cpu that is needed.
1136  * @flags: attributes to set for the ring buffer.
1137  *
1138  * Currently the only flag that is available is the RB_FL_OVERWRITE
1139  * flag. This flag means that the buffer will overwrite old data
1140  * when the buffer wraps. If this flag is not set, the buffer will
1141  * drop data when the tail hits the head.
1142  */
1143 struct ring_buffer *__ring_buffer_alloc(unsigned long size, unsigned flags,
1144                                         struct lock_class_key *key)
1145 {
1146         struct ring_buffer *buffer;
1147         int bsize;
1148         int cpu;
1149
1150         /* keep it in its own cache line */
1151         buffer = kzalloc(ALIGN(sizeof(*buffer), cache_line_size()),
1152                          GFP_KERNEL);
1153         if (!buffer)
1154                 return NULL;
1155
1156         if (!alloc_cpumask_var(&buffer->cpumask, GFP_KERNEL))
1157                 goto fail_free_buffer;
1158
1159         buffer->pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
1160         buffer->flags = flags;
1161         buffer->clock = trace_clock_local;
1162         buffer->reader_lock_key = key;
1163
1164         /* need at least two pages */
1165         if (buffer->pages < 2)
1166                 buffer->pages = 2;
1167
1168         /*
1169          * In case of non-hotplug cpu, if the ring-buffer is allocated
1170          * in early initcall, it will not be notified of secondary cpus.
1171          * In that off case, we need to allocate for all possible cpus.
1172          */
1173 #ifdef CONFIG_HOTPLUG_CPU
1174         get_online_cpus();
1175         cpumask_copy(buffer->cpumask, cpu_online_mask);
1176 #else
1177         cpumask_copy(buffer->cpumask, cpu_possible_mask);
1178 #endif
1179         buffer->cpus = nr_cpu_ids;
1180
1181         bsize = sizeof(void *) * nr_cpu_ids;
1182         buffer->buffers = kzalloc(ALIGN(bsize, cache_line_size()),
1183                                   GFP_KERNEL);
1184         if (!buffer->buffers)
1185                 goto fail_free_cpumask;
1186
1187         for_each_buffer_cpu(buffer, cpu) {
1188                 buffer->buffers[cpu] =
1189                         rb_allocate_cpu_buffer(buffer, cpu);
1190                 if (!buffer->buffers[cpu])
1191                         goto fail_free_buffers;
1192         }
1193
1194 #ifdef CONFIG_HOTPLUG_CPU
1195         buffer->cpu_notify.notifier_call = rb_cpu_notify;
1196         buffer->cpu_notify.priority = 0;
1197         register_cpu_notifier(&buffer->cpu_notify);
1198 #endif
1199
1200         put_online_cpus();
1201         mutex_init(&buffer->mutex);
1202
1203         return buffer;
1204
1205  fail_free_buffers:
1206         for_each_buffer_cpu(buffer, cpu) {
1207                 if (buffer->buffers[cpu])
1208                         rb_free_cpu_buffer(buffer->buffers[cpu]);
1209         }
1210         kfree(buffer->buffers);
1211
1212  fail_free_cpumask:
1213         free_cpumask_var(buffer->cpumask);
1214         put_online_cpus();
1215
1216  fail_free_buffer:
1217         kfree(buffer);
1218         return NULL;
1219 }
1220 EXPORT_SYMBOL_GPL(__ring_buffer_alloc);
1221
1222 /**
1223  * ring_buffer_free - free a ring buffer.
1224  * @buffer: the buffer to free.
1225  */
1226 void
1227 ring_buffer_free(struct ring_buffer *buffer)
1228 {
1229         int cpu;
1230
1231         get_online_cpus();
1232
1233 #ifdef CONFIG_HOTPLUG_CPU
1234         unregister_cpu_notifier(&buffer->cpu_notify);
1235 #endif
1236
1237         for_each_buffer_cpu(buffer, cpu)
1238                 rb_free_cpu_buffer(buffer->buffers[cpu]);
1239
1240         put_online_cpus();
1241
1242         kfree(buffer->buffers);
1243         free_cpumask_var(buffer->cpumask);
1244
1245         kfree(buffer);
1246 }
1247 EXPORT_SYMBOL_GPL(ring_buffer_free);
1248
1249 void ring_buffer_set_clock(struct ring_buffer *buffer,
1250                            u64 (*clock)(void))
1251 {
1252         buffer->clock = clock;
1253 }
1254
1255 static void rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer);
1256
1257 static void
1258 rb_remove_pages(struct ring_buffer_per_cpu *cpu_buffer, unsigned nr_pages)
1259 {
1260         struct buffer_page *bpage;
1261         struct list_head *p;
1262         unsigned i;
1263
1264         raw_spin_lock_irq(&cpu_buffer->reader_lock);
1265         rb_head_page_deactivate(cpu_buffer);
1266
1267         for (i = 0; i < nr_pages; i++) {
1268                 if (RB_WARN_ON(cpu_buffer, list_empty(cpu_buffer->pages)))
1269                         goto out;
1270                 p = cpu_buffer->pages->next;
1271                 bpage = list_entry(p, struct buffer_page, list);
1272                 list_del_init(&bpage->list);
1273                 free_buffer_page(bpage);
1274         }
1275         if (RB_WARN_ON(cpu_buffer, list_empty(cpu_buffer->pages)))
1276                 goto out;
1277
1278         rb_reset_cpu(cpu_buffer);
1279         rb_check_pages(cpu_buffer);
1280
1281 out:
1282         raw_spin_unlock_irq(&cpu_buffer->reader_lock);
1283 }
1284
1285 static void
1286 rb_insert_pages(struct ring_buffer_per_cpu *cpu_buffer,
1287                 struct list_head *pages, unsigned nr_pages)
1288 {
1289         struct buffer_page *bpage;
1290         struct list_head *p;
1291         unsigned i;
1292
1293         raw_spin_lock_irq(&cpu_buffer->reader_lock);
1294         rb_head_page_deactivate(cpu_buffer);
1295
1296         for (i = 0; i < nr_pages; i++) {
1297                 if (RB_WARN_ON(cpu_buffer, list_empty(pages)))
1298                         goto out;
1299                 p = pages->next;
1300                 bpage = list_entry(p, struct buffer_page, list);
1301                 list_del_init(&bpage->list);
1302                 list_add_tail(&bpage->list, cpu_buffer->pages);
1303         }
1304         rb_reset_cpu(cpu_buffer);
1305         rb_check_pages(cpu_buffer);
1306
1307 out:
1308         raw_spin_unlock_irq(&cpu_buffer->reader_lock);
1309 }
1310
1311 /**
1312  * ring_buffer_resize - resize the ring buffer
1313  * @buffer: the buffer to resize.
1314  * @size: the new size.
1315  *
1316  * Minimum size is 2 * BUF_PAGE_SIZE.
1317  *
1318  * Returns -1 on failure.
1319  */
1320 int ring_buffer_resize(struct ring_buffer *buffer, unsigned long size)
1321 {
1322         struct ring_buffer_per_cpu *cpu_buffer;
1323         unsigned nr_pages, rm_pages, new_pages;
1324         struct buffer_page *bpage, *tmp;
1325         unsigned long buffer_size;
1326         LIST_HEAD(pages);
1327         int i, cpu;
1328
1329         /*
1330          * Always succeed at resizing a non-existent buffer:
1331          */
1332         if (!buffer)
1333                 return size;
1334
1335         size = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
1336         size *= BUF_PAGE_SIZE;
1337         buffer_size = buffer->pages * BUF_PAGE_SIZE;
1338
1339         /* we need a minimum of two pages */
1340         if (size < BUF_PAGE_SIZE * 2)
1341                 size = BUF_PAGE_SIZE * 2;
1342
1343         if (size == buffer_size)
1344                 return size;
1345
1346         atomic_inc(&buffer->record_disabled);
1347
1348         /* Make sure all writers are done with this buffer. */
1349         synchronize_sched();
1350
1351         mutex_lock(&buffer->mutex);
1352         get_online_cpus();
1353
1354         nr_pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
1355
1356         if (size < buffer_size) {
1357
1358                 /* easy case, just free pages */
1359                 if (RB_WARN_ON(buffer, nr_pages >= buffer->pages))
1360                         goto out_fail;
1361
1362                 rm_pages = buffer->pages - nr_pages;
1363
1364                 for_each_buffer_cpu(buffer, cpu) {
1365                         cpu_buffer = buffer->buffers[cpu];
1366                         rb_remove_pages(cpu_buffer, rm_pages);
1367                 }
1368                 goto out;
1369         }
1370
1371         /*
1372          * This is a bit more difficult. We only want to add pages
1373          * when we can allocate enough for all CPUs. We do this
1374          * by allocating all the pages and storing them on a local
1375          * link list. If we succeed in our allocation, then we
1376          * add these pages to the cpu_buffers. Otherwise we just free
1377          * them all and return -ENOMEM;
1378          */
1379         if (RB_WARN_ON(buffer, nr_pages <= buffer->pages))
1380                 goto out_fail;
1381
1382         new_pages = nr_pages - buffer->pages;
1383
1384         for_each_buffer_cpu(buffer, cpu) {
1385                 for (i = 0; i < new_pages; i++) {
1386                         struct page *page;
1387                         /*
1388                          * __GFP_NORETRY flag makes sure that the allocation
1389                          * fails gracefully without invoking oom-killer and
1390                          * the system is not destabilized.
1391                          */
1392                         bpage = kzalloc_node(ALIGN(sizeof(*bpage),
1393                                                   cache_line_size()),
1394                                             GFP_KERNEL | __GFP_NORETRY,
1395                                             cpu_to_node(cpu));
1396                         if (!bpage)
1397                                 goto free_pages;
1398                         list_add(&bpage->list, &pages);
1399                         page = alloc_pages_node(cpu_to_node(cpu),
1400                                                 GFP_KERNEL | __GFP_NORETRY, 0);
1401                         if (!page)
1402                                 goto free_pages;
1403                         bpage->page = page_address(page);
1404                         rb_init_page(bpage->page);
1405                 }
1406         }
1407
1408         for_each_buffer_cpu(buffer, cpu) {
1409                 cpu_buffer = buffer->buffers[cpu];
1410                 rb_insert_pages(cpu_buffer, &pages, new_pages);
1411         }
1412
1413         if (RB_WARN_ON(buffer, !list_empty(&pages)))
1414                 goto out_fail;
1415
1416  out:
1417         buffer->pages = nr_pages;
1418         put_online_cpus();
1419         mutex_unlock(&buffer->mutex);
1420
1421         atomic_dec(&buffer->record_disabled);
1422
1423         return size;
1424
1425  free_pages:
1426         list_for_each_entry_safe(bpage, tmp, &pages, list) {
1427                 list_del_init(&bpage->list);
1428                 free_buffer_page(bpage);
1429         }
1430         put_online_cpus();
1431         mutex_unlock(&buffer->mutex);
1432         atomic_dec(&buffer->record_disabled);
1433         return -ENOMEM;
1434
1435         /*
1436          * Something went totally wrong, and we are too paranoid
1437          * to even clean up the mess.
1438          */
1439  out_fail:
1440         put_online_cpus();
1441         mutex_unlock(&buffer->mutex);
1442         atomic_dec(&buffer->record_disabled);
1443         return -1;
1444 }
1445 EXPORT_SYMBOL_GPL(ring_buffer_resize);
1446
1447 void ring_buffer_change_overwrite(struct ring_buffer *buffer, int val)
1448 {
1449         mutex_lock(&buffer->mutex);
1450         if (val)
1451                 buffer->flags |= RB_FL_OVERWRITE;
1452         else
1453                 buffer->flags &= ~RB_FL_OVERWRITE;
1454         mutex_unlock(&buffer->mutex);
1455 }
1456 EXPORT_SYMBOL_GPL(ring_buffer_change_overwrite);
1457
1458 static inline void *
1459 __rb_data_page_index(struct buffer_data_page *bpage, unsigned index)
1460 {
1461         return bpage->data + index;
1462 }
1463
1464 static inline void *__rb_page_index(struct buffer_page *bpage, unsigned index)
1465 {
1466         return bpage->page->data + index;
1467 }
1468
1469 static inline struct ring_buffer_event *
1470 rb_reader_event(struct ring_buffer_per_cpu *cpu_buffer)
1471 {
1472         return __rb_page_index(cpu_buffer->reader_page,
1473                                cpu_buffer->reader_page->read);
1474 }
1475
1476 static inline struct ring_buffer_event *
1477 rb_iter_head_event(struct ring_buffer_iter *iter)
1478 {
1479         return __rb_page_index(iter->head_page, iter->head);
1480 }
1481
1482 static inline unsigned long rb_page_write(struct buffer_page *bpage)
1483 {
1484         return local_read(&bpage->write) & RB_WRITE_MASK;
1485 }
1486
1487 static inline unsigned rb_page_commit(struct buffer_page *bpage)
1488 {
1489         return local_read(&bpage->page->commit);
1490 }
1491
1492 static inline unsigned long rb_page_entries(struct buffer_page *bpage)
1493 {
1494         return local_read(&bpage->entries) & RB_WRITE_MASK;
1495 }
1496
1497 /* Size is determined by what has been committed */
1498 static inline unsigned rb_page_size(struct buffer_page *bpage)
1499 {
1500         return rb_page_commit(bpage);
1501 }
1502
1503 static inline unsigned
1504 rb_commit_index(struct ring_buffer_per_cpu *cpu_buffer)
1505 {
1506         return rb_page_commit(cpu_buffer->commit_page);
1507 }
1508
1509 static inline unsigned
1510 rb_event_index(struct ring_buffer_event *event)
1511 {
1512         unsigned long addr = (unsigned long)event;
1513
1514         return (addr & ~PAGE_MASK) - BUF_PAGE_HDR_SIZE;
1515 }
1516
1517 static inline int
1518 rb_event_is_commit(struct ring_buffer_per_cpu *cpu_buffer,
1519                    struct ring_buffer_event *event)
1520 {
1521         unsigned long addr = (unsigned long)event;
1522         unsigned long index;
1523
1524         index = rb_event_index(event);
1525         addr &= PAGE_MASK;
1526
1527         return cpu_buffer->commit_page->page == (void *)addr &&
1528                 rb_commit_index(cpu_buffer) == index;
1529 }
1530
1531 static void
1532 rb_set_commit_to_write(struct ring_buffer_per_cpu *cpu_buffer)
1533 {
1534         unsigned long max_count;
1535
1536         /*
1537          * We only race with interrupts and NMIs on this CPU.
1538          * If we own the commit event, then we can commit
1539          * all others that interrupted us, since the interruptions
1540          * are in stack format (they finish before they come
1541          * back to us). This allows us to do a simple loop to
1542          * assign the commit to the tail.
1543          */
1544  again:
1545         max_count = cpu_buffer->buffer->pages * 100;
1546
1547         while (cpu_buffer->commit_page != cpu_buffer->tail_page) {
1548                 if (RB_WARN_ON(cpu_buffer, !(--max_count)))
1549                         return;
1550                 if (RB_WARN_ON(cpu_buffer,
1551                                rb_is_reader_page(cpu_buffer->tail_page)))
1552                         return;
1553                 local_set(&cpu_buffer->commit_page->page->commit,
1554                           rb_page_write(cpu_buffer->commit_page));
1555                 rb_inc_page(cpu_buffer, &cpu_buffer->commit_page);
1556                 cpu_buffer->write_stamp =
1557                         cpu_buffer->commit_page->page->time_stamp;
1558                 /* add barrier to keep gcc from optimizing too much */
1559                 barrier();
1560         }
1561         while (rb_commit_index(cpu_buffer) !=
1562                rb_page_write(cpu_buffer->commit_page)) {
1563
1564                 local_set(&cpu_buffer->commit_page->page->commit,
1565                           rb_page_write(cpu_buffer->commit_page));
1566                 RB_WARN_ON(cpu_buffer,
1567                            local_read(&cpu_buffer->commit_page->page->commit) &
1568                            ~RB_WRITE_MASK);
1569                 barrier();
1570         }
1571
1572         /* again, keep gcc from optimizing */
1573         barrier();
1574
1575         /*
1576          * If an interrupt came in just after the first while loop
1577          * and pushed the tail page forward, we will be left with
1578          * a dangling commit that will never go forward.
1579          */
1580         if (unlikely(cpu_buffer->commit_page != cpu_buffer->tail_page))
1581                 goto again;
1582 }
1583
1584 static void rb_reset_reader_page(struct ring_buffer_per_cpu *cpu_buffer)
1585 {
1586         cpu_buffer->read_stamp = cpu_buffer->reader_page->page->time_stamp;
1587         cpu_buffer->reader_page->read = 0;
1588 }
1589
1590 static void rb_inc_iter(struct ring_buffer_iter *iter)
1591 {
1592         struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
1593
1594         /*
1595          * The iterator could be on the reader page (it starts there).
1596          * But the head could have moved, since the reader was
1597          * found. Check for this case and assign the iterator
1598          * to the head page instead of next.
1599          */
1600         if (iter->head_page == cpu_buffer->reader_page)
1601                 iter->head_page = rb_set_head_page(cpu_buffer);
1602         else
1603                 rb_inc_page(cpu_buffer, &iter->head_page);
1604
1605         iter->read_stamp = iter->head_page->page->time_stamp;
1606         iter->head = 0;
1607 }
1608
1609 /* Slow path, do not inline */
1610 static noinline struct ring_buffer_event *
1611 rb_add_time_stamp(struct ring_buffer_event *event, u64 delta)
1612 {
1613         event->type_len = RINGBUF_TYPE_TIME_EXTEND;
1614
1615         /* Not the first event on the page? */
1616         if (rb_event_index(event)) {
1617                 event->time_delta = delta & TS_MASK;
1618                 event->array[0] = delta >> TS_SHIFT;
1619         } else {
1620                 /* nope, just zero it */
1621                 event->time_delta = 0;
1622                 event->array[0] = 0;
1623         }
1624
1625         return skip_time_extend(event);
1626 }
1627
1628 /**
1629  * ring_buffer_update_event - update event type and data
1630  * @event: the even to update
1631  * @type: the type of event
1632  * @length: the size of the event field in the ring buffer
1633  *
1634  * Update the type and data fields of the event. The length
1635  * is the actual size that is written to the ring buffer,
1636  * and with this, we can determine what to place into the
1637  * data field.
1638  */
1639 static void
1640 rb_update_event(struct ring_buffer_per_cpu *cpu_buffer,
1641                 struct ring_buffer_event *event, unsigned length,
1642                 int add_timestamp, u64 delta)
1643 {
1644         /* Only a commit updates the timestamp */
1645         if (unlikely(!rb_event_is_commit(cpu_buffer, event)))
1646                 delta = 0;
1647
1648         /*
1649          * If we need to add a timestamp, then we
1650          * add it to the start of the resevered space.
1651          */
1652         if (unlikely(add_timestamp)) {
1653                 event = rb_add_time_stamp(event, delta);
1654                 length -= RB_LEN_TIME_EXTEND;
1655                 delta = 0;
1656         }
1657
1658         event->time_delta = delta;
1659         length -= RB_EVNT_HDR_SIZE;
1660         if (length > RB_MAX_SMALL_DATA || RB_FORCE_8BYTE_ALIGNMENT) {
1661                 event->type_len = 0;
1662                 event->array[0] = length;
1663         } else
1664                 event->type_len = DIV_ROUND_UP(length, RB_ALIGNMENT);
1665 }
1666
1667 /*
1668  * rb_handle_head_page - writer hit the head page
1669  *
1670  * Returns: +1 to retry page
1671  *           0 to continue
1672  *          -1 on error
1673  */
1674 static int
1675 rb_handle_head_page(struct ring_buffer_per_cpu *cpu_buffer,
1676                     struct buffer_page *tail_page,
1677                     struct buffer_page *next_page)
1678 {
1679         struct buffer_page *new_head;
1680         int entries;
1681         int type;
1682         int ret;
1683
1684         entries = rb_page_entries(next_page);
1685
1686         /*
1687          * The hard part is here. We need to move the head
1688          * forward, and protect against both readers on
1689          * other CPUs and writers coming in via interrupts.
1690          */
1691         type = rb_head_page_set_update(cpu_buffer, next_page, tail_page,
1692                                        RB_PAGE_HEAD);
1693
1694         /*
1695          * type can be one of four:
1696          *  NORMAL - an interrupt already moved it for us
1697          *  HEAD   - we are the first to get here.
1698          *  UPDATE - we are the interrupt interrupting
1699          *           a current move.
1700          *  MOVED  - a reader on another CPU moved the next
1701          *           pointer to its reader page. Give up
1702          *           and try again.
1703          */
1704
1705         switch (type) {
1706         case RB_PAGE_HEAD:
1707                 /*
1708                  * We changed the head to UPDATE, thus
1709                  * it is our responsibility to update
1710                  * the counters.
1711                  */
1712                 local_add(entries, &cpu_buffer->overrun);
1713                 local_sub(BUF_PAGE_SIZE, &cpu_buffer->entries_bytes);
1714
1715                 /*
1716                  * The entries will be zeroed out when we move the
1717                  * tail page.
1718                  */
1719
1720                 /* still more to do */
1721                 break;
1722
1723         case RB_PAGE_UPDATE:
1724                 /*
1725                  * This is an interrupt that interrupt the
1726                  * previous update. Still more to do.
1727                  */
1728                 break;
1729         case RB_PAGE_NORMAL:
1730                 /*
1731                  * An interrupt came in before the update
1732                  * and processed this for us.
1733                  * Nothing left to do.
1734                  */
1735                 return 1;
1736         case RB_PAGE_MOVED:
1737                 /*
1738                  * The reader is on another CPU and just did
1739                  * a swap with our next_page.
1740                  * Try again.
1741                  */
1742                 return 1;
1743         default:
1744                 RB_WARN_ON(cpu_buffer, 1); /* WTF??? */
1745                 return -1;
1746         }
1747
1748         /*
1749          * Now that we are here, the old head pointer is
1750          * set to UPDATE. This will keep the reader from
1751          * swapping the head page with the reader page.
1752          * The reader (on another CPU) will spin till
1753          * we are finished.
1754          *
1755          * We just need to protect against interrupts
1756          * doing the job. We will set the next pointer
1757          * to HEAD. After that, we set the old pointer
1758          * to NORMAL, but only if it was HEAD before.
1759          * otherwise we are an interrupt, and only
1760          * want the outer most commit to reset it.
1761          */
1762         new_head = next_page;
1763         rb_inc_page(cpu_buffer, &new_head);
1764
1765         ret = rb_head_page_set_head(cpu_buffer, new_head, next_page,
1766                                     RB_PAGE_NORMAL);
1767
1768         /*
1769          * Valid returns are:
1770          *  HEAD   - an interrupt came in and already set it.
1771          *  NORMAL - One of two things:
1772          *            1) We really set it.
1773          *            2) A bunch of interrupts came in and moved
1774          *               the page forward again.
1775          */
1776         switch (ret) {
1777         case RB_PAGE_HEAD:
1778         case RB_PAGE_NORMAL:
1779                 /* OK */
1780                 break;
1781         default:
1782                 RB_WARN_ON(cpu_buffer, 1);
1783                 return -1;
1784         }
1785
1786         /*
1787          * It is possible that an interrupt came in,
1788          * set the head up, then more interrupts came in
1789          * and moved it again. When we get back here,
1790          * the page would have been set to NORMAL but we
1791          * just set it back to HEAD.
1792          *
1793          * How do you detect this? Well, if that happened
1794          * the tail page would have moved.
1795          */
1796         if (ret == RB_PAGE_NORMAL) {
1797                 /*
1798                  * If the tail had moved passed next, then we need
1799                  * to reset the pointer.
1800                  */
1801                 if (cpu_buffer->tail_page != tail_page &&
1802                     cpu_buffer->tail_page != next_page)
1803                         rb_head_page_set_normal(cpu_buffer, new_head,
1804                                                 next_page,
1805                                                 RB_PAGE_HEAD);
1806         }
1807
1808         /*
1809          * If this was the outer most commit (the one that
1810          * changed the original pointer from HEAD to UPDATE),
1811          * then it is up to us to reset it to NORMAL.
1812          */
1813         if (type == RB_PAGE_HEAD) {
1814                 ret = rb_head_page_set_normal(cpu_buffer, next_page,
1815                                               tail_page,
1816                                               RB_PAGE_UPDATE);
1817                 if (RB_WARN_ON(cpu_buffer,
1818                                ret != RB_PAGE_UPDATE))
1819                         return -1;
1820         }
1821
1822         return 0;
1823 }
1824
1825 static unsigned rb_calculate_event_length(unsigned length)
1826 {
1827         struct ring_buffer_event event; /* Used only for sizeof array */
1828
1829         /* zero length can cause confusions */
1830         if (!length)
1831                 length = 1;
1832
1833         if (length > RB_MAX_SMALL_DATA || RB_FORCE_8BYTE_ALIGNMENT)
1834                 length += sizeof(event.array[0]);
1835
1836         length += RB_EVNT_HDR_SIZE;
1837         length = ALIGN(length, RB_ARCH_ALIGNMENT);
1838
1839         return length;
1840 }
1841
1842 static inline void
1843 rb_reset_tail(struct ring_buffer_per_cpu *cpu_buffer,
1844               struct buffer_page *tail_page,
1845               unsigned long tail, unsigned long length)
1846 {
1847         struct ring_buffer_event *event;
1848
1849         /*
1850          * Only the event that crossed the page boundary
1851          * must fill the old tail_page with padding.
1852          */
1853         if (tail >= BUF_PAGE_SIZE) {
1854                 /*
1855                  * If the page was filled, then we still need
1856                  * to update the real_end. Reset it to zero
1857                  * and the reader will ignore it.
1858                  */
1859                 if (tail == BUF_PAGE_SIZE)
1860                         tail_page->real_end = 0;
1861
1862                 local_sub(length, &tail_page->write);
1863                 return;
1864         }
1865
1866         event = __rb_page_index(tail_page, tail);
1867         kmemcheck_annotate_bitfield(event, bitfield);
1868
1869         /* account for padding bytes */
1870         local_add(BUF_PAGE_SIZE - tail, &cpu_buffer->entries_bytes);
1871
1872         /*
1873          * Save the original length to the meta data.
1874          * This will be used by the reader to add lost event
1875          * counter.
1876          */
1877         tail_page->real_end = tail;
1878
1879         /*
1880          * If this event is bigger than the minimum size, then
1881          * we need to be careful that we don't subtract the
1882          * write counter enough to allow another writer to slip
1883          * in on this page.
1884          * We put in a discarded commit instead, to make sure
1885          * that this space is not used again.
1886          *
1887          * If we are less than the minimum size, we don't need to
1888          * worry about it.
1889          */
1890         if (tail > (BUF_PAGE_SIZE - RB_EVNT_MIN_SIZE)) {
1891                 /* No room for any events */
1892
1893                 /* Mark the rest of the page with padding */
1894                 rb_event_set_padding(event);
1895
1896                 /* Set the write back to the previous setting */
1897                 local_sub(length, &tail_page->write);
1898                 return;
1899         }
1900
1901         /* Put in a discarded event */
1902         event->array[0] = (BUF_PAGE_SIZE - tail) - RB_EVNT_HDR_SIZE;
1903         event->type_len = RINGBUF_TYPE_PADDING;
1904         /* time delta must be non zero */
1905         event->time_delta = 1;
1906
1907         /* Set write to end of buffer */
1908         length = (tail + length) - BUF_PAGE_SIZE;
1909         local_sub(length, &tail_page->write);
1910 }
1911
1912 /*
1913  * This is the slow path, force gcc not to inline it.
1914  */
1915 static noinline struct ring_buffer_event *
1916 rb_move_tail(struct ring_buffer_per_cpu *cpu_buffer,
1917              unsigned long length, unsigned long tail,
1918              struct buffer_page *tail_page, u64 ts)
1919 {
1920         struct buffer_page *commit_page = cpu_buffer->commit_page;
1921         struct ring_buffer *buffer = cpu_buffer->buffer;
1922         struct buffer_page *next_page;
1923         int ret;
1924
1925         next_page = tail_page;
1926
1927         rb_inc_page(cpu_buffer, &next_page);
1928
1929         /*
1930          * If for some reason, we had an interrupt storm that made
1931          * it all the way around the buffer, bail, and warn
1932          * about it.
1933          */
1934         if (unlikely(next_page == commit_page)) {
1935                 local_inc(&cpu_buffer->commit_overrun);
1936                 goto out_reset;
1937         }
1938
1939         /*
1940          * This is where the fun begins!
1941          *
1942          * We are fighting against races between a reader that
1943          * could be on another CPU trying to swap its reader
1944          * page with the buffer head.
1945          *
1946          * We are also fighting against interrupts coming in and
1947          * moving the head or tail on us as well.
1948          *
1949          * If the next page is the head page then we have filled
1950          * the buffer, unless the commit page is still on the
1951          * reader page.
1952          */
1953         if (rb_is_head_page(cpu_buffer, next_page, &tail_page->list)) {
1954
1955                 /*
1956                  * If the commit is not on the reader page, then
1957                  * move the header page.
1958                  */
1959                 if (!rb_is_reader_page(cpu_buffer->commit_page)) {
1960                         /*
1961                          * If we are not in overwrite mode,
1962                          * this is easy, just stop here.
1963                          */
1964                         if (!(buffer->flags & RB_FL_OVERWRITE))
1965                                 goto out_reset;
1966
1967                         ret = rb_handle_head_page(cpu_buffer,
1968                                                   tail_page,
1969                                                   next_page);
1970                         if (ret < 0)
1971                                 goto out_reset;
1972                         if (ret)
1973                                 goto out_again;
1974                 } else {
1975                         /*
1976                          * We need to be careful here too. The
1977                          * commit page could still be on the reader
1978                          * page. We could have a small buffer, and
1979                          * have filled up the buffer with events
1980                          * from interrupts and such, and wrapped.
1981                          *
1982                          * Note, if the tail page is also the on the
1983                          * reader_page, we let it move out.
1984                          */
1985                         if (unlikely((cpu_buffer->commit_page !=
1986                                       cpu_buffer->tail_page) &&
1987                                      (cpu_buffer->commit_page ==
1988                                       cpu_buffer->reader_page))) {
1989                                 local_inc(&cpu_buffer->commit_overrun);
1990                                 goto out_reset;
1991                         }
1992                 }
1993         }
1994
1995         ret = rb_tail_page_update(cpu_buffer, tail_page, next_page);
1996         if (ret) {
1997                 /*
1998                  * Nested commits always have zero deltas, so
1999                  * just reread the time stamp
2000                  */
2001                 ts = rb_time_stamp(buffer);
2002                 next_page->page->time_stamp = ts;
2003         }
2004
2005  out_again:
2006
2007         rb_reset_tail(cpu_buffer, tail_page, tail, length);
2008
2009         /* fail and let the caller try again */
2010         return ERR_PTR(-EAGAIN);
2011
2012  out_reset:
2013         /* reset write */
2014         rb_reset_tail(cpu_buffer, tail_page, tail, length);
2015
2016         return NULL;
2017 }
2018
2019 static struct ring_buffer_event *
2020 __rb_reserve_next(struct ring_buffer_per_cpu *cpu_buffer,
2021                   unsigned long length, u64 ts,
2022                   u64 delta, int add_timestamp)
2023 {
2024         struct buffer_page *tail_page;
2025         struct ring_buffer_event *event;
2026         unsigned long tail, write;
2027
2028         /*
2029          * If the time delta since the last event is too big to
2030          * hold in the time field of the event, then we append a
2031          * TIME EXTEND event ahead of the data event.
2032          */
2033         if (unlikely(add_timestamp))
2034                 length += RB_LEN_TIME_EXTEND;
2035
2036         tail_page = cpu_buffer->tail_page;
2037         write = local_add_return(length, &tail_page->write);
2038
2039         /* set write to only the index of the write */
2040         write &= RB_WRITE_MASK;
2041         tail = write - length;
2042
2043         /*
2044          * If this is the first commit on the page, then it has the same
2045          * timestamp as the page itself.
2046          */
2047         if (!tail)
2048                 delta = 0;
2049
2050         /* See if we shot pass the end of this buffer page */
2051         if (unlikely(write > BUF_PAGE_SIZE))
2052                 return rb_move_tail(cpu_buffer, length, tail,
2053                                     tail_page, ts);
2054
2055         /* We reserved something on the buffer */
2056
2057         event = __rb_page_index(tail_page, tail);
2058         kmemcheck_annotate_bitfield(event, bitfield);
2059         rb_update_event(cpu_buffer, event, length, add_timestamp, delta);
2060
2061         local_inc(&tail_page->entries);
2062
2063         /*
2064          * If this is the first commit on the page, then update
2065          * its timestamp.
2066          */
2067         if (!tail)
2068                 tail_page->page->time_stamp = ts;
2069
2070         /* account for these added bytes */
2071         local_add(length, &cpu_buffer->entries_bytes);
2072
2073         return event;
2074 }
2075
2076 static inline int
2077 rb_try_to_discard(struct ring_buffer_per_cpu *cpu_buffer,
2078                   struct ring_buffer_event *event)
2079 {
2080         unsigned long new_index, old_index;
2081         struct buffer_page *bpage;
2082         unsigned long index;
2083         unsigned long addr;
2084
2085         new_index = rb_event_index(event);
2086         old_index = new_index + rb_event_ts_length(event);
2087         addr = (unsigned long)event;
2088         addr &= PAGE_MASK;
2089
2090         bpage = cpu_buffer->tail_page;
2091
2092         if (bpage->page == (void *)addr && rb_page_write(bpage) == old_index) {
2093                 unsigned long write_mask =
2094                         local_read(&bpage->write) & ~RB_WRITE_MASK;
2095                 unsigned long event_length = rb_event_length(event);
2096                 /*
2097                  * This is on the tail page. It is possible that
2098                  * a write could come in and move the tail page
2099                  * and write to the next page. That is fine
2100                  * because we just shorten what is on this page.
2101                  */
2102                 old_index += write_mask;
2103                 new_index += write_mask;
2104                 index = local_cmpxchg(&bpage->write, old_index, new_index);
2105                 if (index == old_index) {
2106                         /* update counters */
2107                         local_sub(event_length, &cpu_buffer->entries_bytes);
2108                         return 1;
2109                 }
2110         }
2111
2112         /* could not discard */
2113         return 0;
2114 }
2115
2116 static void rb_start_commit(struct ring_buffer_per_cpu *cpu_buffer)
2117 {
2118         local_inc(&cpu_buffer->committing);
2119         local_inc(&cpu_buffer->commits);
2120 }
2121
2122 static inline void rb_end_commit(struct ring_buffer_per_cpu *cpu_buffer)
2123 {
2124         unsigned long commits;
2125
2126         if (RB_WARN_ON(cpu_buffer,
2127                        !local_read(&cpu_buffer->committing)))
2128                 return;
2129
2130  again:
2131         commits = local_read(&cpu_buffer->commits);
2132         /* synchronize with interrupts */
2133         barrier();
2134         if (local_read(&cpu_buffer->committing) == 1)
2135                 rb_set_commit_to_write(cpu_buffer);
2136
2137         local_dec(&cpu_buffer->committing);
2138
2139         /* synchronize with interrupts */
2140         barrier();
2141
2142         /*
2143          * Need to account for interrupts coming in between the
2144          * updating of the commit page and the clearing of the
2145          * committing counter.
2146          */
2147         if (unlikely(local_read(&cpu_buffer->commits) != commits) &&
2148             !local_read(&cpu_buffer->committing)) {
2149                 local_inc(&cpu_buffer->committing);
2150                 goto again;
2151         }
2152 }
2153
2154 static struct ring_buffer_event *
2155 rb_reserve_next_event(struct ring_buffer *buffer,
2156                       struct ring_buffer_per_cpu *cpu_buffer,
2157                       unsigned long length)
2158 {
2159         struct ring_buffer_event *event;
2160         u64 ts, delta;
2161         int nr_loops = 0;
2162         int add_timestamp;
2163         u64 diff;
2164
2165         rb_start_commit(cpu_buffer);
2166
2167 #ifdef CONFIG_RING_BUFFER_ALLOW_SWAP
2168         /*
2169          * Due to the ability to swap a cpu buffer from a buffer
2170          * it is possible it was swapped before we committed.
2171          * (committing stops a swap). We check for it here and
2172          * if it happened, we have to fail the write.
2173          */
2174         barrier();
2175         if (unlikely(ACCESS_ONCE(cpu_buffer->buffer) != buffer)) {
2176                 local_dec(&cpu_buffer->committing);
2177                 local_dec(&cpu_buffer->commits);
2178                 return NULL;
2179         }
2180 #endif
2181
2182         length = rb_calculate_event_length(length);
2183  again:
2184         add_timestamp = 0;
2185         delta = 0;
2186
2187         /*
2188          * We allow for interrupts to reenter here and do a trace.
2189          * If one does, it will cause this original code to loop
2190          * back here. Even with heavy interrupts happening, this
2191          * should only happen a few times in a row. If this happens
2192          * 1000 times in a row, there must be either an interrupt
2193          * storm or we have something buggy.
2194          * Bail!
2195          */
2196         if (RB_WARN_ON(cpu_buffer, ++nr_loops > 1000))
2197                 goto out_fail;
2198
2199         ts = rb_time_stamp(cpu_buffer->buffer);
2200         diff = ts - cpu_buffer->write_stamp;
2201
2202         /* make sure this diff is calculated here */
2203         barrier();
2204
2205         /* Did the write stamp get updated already? */
2206         if (likely(ts >= cpu_buffer->write_stamp)) {
2207                 delta = diff;
2208                 if (unlikely(test_time_stamp(delta))) {
2209                         int local_clock_stable = 1;
2210 #ifdef CONFIG_HAVE_UNSTABLE_SCHED_CLOCK
2211                         local_clock_stable = sched_clock_stable;
2212 #endif
2213                         WARN_ONCE(delta > (1ULL << 59),
2214                                   KERN_WARNING "Delta way too big! %llu ts=%llu write stamp = %llu\n%s",
2215                                   (unsigned long long)delta,
2216                                   (unsigned long long)ts,
2217                                   (unsigned long long)cpu_buffer->write_stamp,
2218                                   local_clock_stable ? "" :
2219                                   "If you just came from a suspend/resume,\n"
2220                                   "please switch to the trace global clock:\n"
2221                                   "  echo global > /sys/kernel/debug/tracing/trace_clock\n");
2222                         add_timestamp = 1;
2223                 }
2224         }
2225
2226         event = __rb_reserve_next(cpu_buffer, length, ts,
2227                                   delta, add_timestamp);
2228         if (unlikely(PTR_ERR(event) == -EAGAIN))
2229                 goto again;
2230
2231         if (!event)
2232                 goto out_fail;
2233
2234         return event;
2235
2236  out_fail:
2237         rb_end_commit(cpu_buffer);
2238         return NULL;
2239 }
2240
2241 #ifdef CONFIG_TRACING
2242
2243 #define TRACE_RECURSIVE_DEPTH 16
2244
2245 /* Keep this code out of the fast path cache */
2246 static noinline void trace_recursive_fail(void)
2247 {
2248         /* Disable all tracing before we do anything else */
2249         tracing_off_permanent();
2250
2251         printk_once(KERN_WARNING "Tracing recursion: depth[%ld]:"
2252                     "HC[%lu]:SC[%lu]:NMI[%lu]\n",
2253                     trace_recursion_buffer(),
2254                     hardirq_count() >> HARDIRQ_SHIFT,
2255                     softirq_count() >> SOFTIRQ_SHIFT,
2256                     in_nmi());
2257
2258         WARN_ON_ONCE(1);
2259 }
2260
2261 static inline int trace_recursive_lock(void)
2262 {
2263         trace_recursion_inc();
2264
2265         if (likely(trace_recursion_buffer() < TRACE_RECURSIVE_DEPTH))
2266                 return 0;
2267
2268         trace_recursive_fail();
2269
2270         return -1;
2271 }
2272
2273 static inline void trace_recursive_unlock(void)
2274 {
2275         WARN_ON_ONCE(!trace_recursion_buffer());
2276
2277         trace_recursion_dec();
2278 }
2279
2280 #else
2281
2282 #define trace_recursive_lock()          (0)
2283 #define trace_recursive_unlock()        do { } while (0)
2284
2285 #endif
2286
2287 /**
2288  * ring_buffer_lock_reserve - reserve a part of the buffer
2289  * @buffer: the ring buffer to reserve from
2290  * @length: the length of the data to reserve (excluding event header)
2291  *
2292  * Returns a reseverd event on the ring buffer to copy directly to.
2293  * The user of this interface will need to get the body to write into
2294  * and can use the ring_buffer_event_data() interface.
2295  *
2296  * The length is the length of the data needed, not the event length
2297  * which also includes the event header.
2298  *
2299  * Must be paired with ring_buffer_unlock_commit, unless NULL is returned.
2300  * If NULL is returned, then nothing has been allocated or locked.
2301  */
2302 struct ring_buffer_event *
2303 ring_buffer_lock_reserve(struct ring_buffer *buffer, unsigned long length)
2304 {
2305         struct ring_buffer_per_cpu *cpu_buffer;
2306         struct ring_buffer_event *event;
2307         int cpu;
2308
2309         if (ring_buffer_flags != RB_BUFFERS_ON)
2310                 return NULL;
2311
2312         /* If we are tracing schedule, we don't want to recurse */
2313         preempt_disable_notrace();
2314
2315         if (atomic_read(&buffer->record_disabled))
2316                 goto out_nocheck;
2317
2318         if (trace_recursive_lock())
2319                 goto out_nocheck;
2320
2321         cpu = raw_smp_processor_id();
2322
2323         if (!cpumask_test_cpu(cpu, buffer->cpumask))
2324                 goto out;
2325
2326         cpu_buffer = buffer->buffers[cpu];
2327
2328         if (atomic_read(&cpu_buffer->record_disabled))
2329                 goto out;
2330
2331         if (length > BUF_MAX_DATA_SIZE)
2332                 goto out;
2333
2334         event = rb_reserve_next_event(buffer, cpu_buffer, length);
2335         if (!event)
2336                 goto out;
2337
2338         return event;
2339
2340  out:
2341         trace_recursive_unlock();
2342
2343  out_nocheck:
2344         preempt_enable_notrace();
2345         return NULL;
2346 }
2347 EXPORT_SYMBOL_GPL(ring_buffer_lock_reserve);
2348
2349 static void
2350 rb_update_write_stamp(struct ring_buffer_per_cpu *cpu_buffer,
2351                       struct ring_buffer_event *event)
2352 {
2353         u64 delta;
2354
2355         /*
2356          * The event first in the commit queue updates the
2357          * time stamp.
2358          */
2359         if (rb_event_is_commit(cpu_buffer, event)) {
2360                 /*
2361                  * A commit event that is first on a page
2362                  * updates the write timestamp with the page stamp
2363                  */
2364                 if (!rb_event_index(event))
2365                         cpu_buffer->write_stamp =
2366                                 cpu_buffer->commit_page->page->time_stamp;
2367                 else if (event->type_len == RINGBUF_TYPE_TIME_EXTEND) {
2368                         delta = event->array[0];
2369                         delta <<= TS_SHIFT;
2370                         delta += event->time_delta;
2371                         cpu_buffer->write_stamp += delta;
2372                 } else
2373                         cpu_buffer->write_stamp += event->time_delta;
2374         }
2375 }
2376
2377 static void rb_commit(struct ring_buffer_per_cpu *cpu_buffer,
2378                       struct ring_buffer_event *event)
2379 {
2380         local_inc(&cpu_buffer->entries);
2381         rb_update_write_stamp(cpu_buffer, event);
2382         rb_end_commit(cpu_buffer);
2383 }
2384
2385 /**
2386  * ring_buffer_unlock_commit - commit a reserved
2387  * @buffer: The buffer to commit to
2388  * @event: The event pointer to commit.
2389  *
2390  * This commits the data to the ring buffer, and releases any locks held.
2391  *
2392  * Must be paired with ring_buffer_lock_reserve.
2393  */
2394 int ring_buffer_unlock_commit(struct ring_buffer *buffer,
2395                               struct ring_buffer_event *event)
2396 {
2397         struct ring_buffer_per_cpu *cpu_buffer;
2398         int cpu = raw_smp_processor_id();
2399
2400         cpu_buffer = buffer->buffers[cpu];
2401
2402         rb_commit(cpu_buffer, event);
2403
2404         trace_recursive_unlock();
2405
2406         preempt_enable_notrace();
2407
2408         return 0;
2409 }
2410 EXPORT_SYMBOL_GPL(ring_buffer_unlock_commit);
2411
2412 static inline void rb_event_discard(struct ring_buffer_event *event)
2413 {
2414         if (event->type_len == RINGBUF_TYPE_TIME_EXTEND)
2415                 event = skip_time_extend(event);
2416
2417         /* array[0] holds the actual length for the discarded event */
2418         event->array[0] = rb_event_data_length(event) - RB_EVNT_HDR_SIZE;
2419         event->type_len = RINGBUF_TYPE_PADDING;
2420         /* time delta must be non zero */
2421         if (!event->time_delta)
2422                 event->time_delta = 1;
2423 }
2424
2425 /*
2426  * Decrement the entries to the page that an event is on.
2427  * The event does not even need to exist, only the pointer
2428  * to the page it is on. This may only be called before the commit
2429  * takes place.
2430  */
2431 static inline void
2432 rb_decrement_entry(struct ring_buffer_per_cpu *cpu_buffer,
2433                    struct ring_buffer_event *event)
2434 {
2435         unsigned long addr = (unsigned long)event;
2436         struct buffer_page *bpage = cpu_buffer->commit_page;
2437         struct buffer_page *start;
2438
2439         addr &= PAGE_MASK;
2440
2441         /* Do the likely case first */
2442         if (likely(bpage->page == (void *)addr)) {
2443                 local_dec(&bpage->entries);
2444                 return;
2445         }
2446
2447         /*
2448          * Because the commit page may be on the reader page we
2449          * start with the next page and check the end loop there.
2450          */
2451         rb_inc_page(cpu_buffer, &bpage);
2452         start = bpage;
2453         do {
2454                 if (bpage->page == (void *)addr) {
2455                         local_dec(&bpage->entries);
2456                         return;
2457                 }
2458                 rb_inc_page(cpu_buffer, &bpage);
2459         } while (bpage != start);
2460
2461         /* commit not part of this buffer?? */
2462         RB_WARN_ON(cpu_buffer, 1);
2463 }
2464
2465 /**
2466  * ring_buffer_commit_discard - discard an event that has not been committed
2467  * @buffer: the ring buffer
2468  * @event: non committed event to discard
2469  *
2470  * Sometimes an event that is in the ring buffer needs to be ignored.
2471  * This function lets the user discard an event in the ring buffer
2472  * and then that event will not be read later.
2473  *
2474  * This function only works if it is called before the the item has been
2475  * committed. It will try to free the event from the ring buffer
2476  * if another event has not been added behind it.
2477  *
2478  * If another event has been added behind it, it will set the event
2479  * up as discarded, and perform the commit.
2480  *
2481  * If this function is called, do not call ring_buffer_unlock_commit on
2482  * the event.
2483  */
2484 void ring_buffer_discard_commit(struct ring_buffer *buffer,
2485                                 struct ring_buffer_event *event)
2486 {
2487         struct ring_buffer_per_cpu *cpu_buffer;
2488         int cpu;
2489
2490         /* The event is discarded regardless */
2491         rb_event_discard(event);
2492
2493         cpu = smp_processor_id();
2494         cpu_buffer = buffer->buffers[cpu];
2495
2496         /*
2497          * This must only be called if the event has not been
2498          * committed yet. Thus we can assume that preemption
2499          * is still disabled.
2500          */
2501         RB_WARN_ON(buffer, !local_read(&cpu_buffer->committing));
2502
2503         rb_decrement_entry(cpu_buffer, event);
2504         if (rb_try_to_discard(cpu_buffer, event))
2505                 goto out;
2506
2507         /*
2508          * The commit is still visible by the reader, so we
2509          * must still update the timestamp.
2510          */
2511         rb_update_write_stamp(cpu_buffer, event);
2512  out:
2513         rb_end_commit(cpu_buffer);
2514
2515         trace_recursive_unlock();
2516
2517         preempt_enable_notrace();
2518
2519 }
2520 EXPORT_SYMBOL_GPL(ring_buffer_discard_commit);
2521
2522 /**
2523  * ring_buffer_write - write data to the buffer without reserving
2524  * @buffer: The ring buffer to write to.
2525  * @length: The length of the data being written (excluding the event header)
2526  * @data: The data to write to the buffer.
2527  *
2528  * This is like ring_buffer_lock_reserve and ring_buffer_unlock_commit as
2529  * one function. If you already have the data to write to the buffer, it
2530  * may be easier to simply call this function.
2531  *
2532  * Note, like ring_buffer_lock_reserve, the length is the length of the data
2533  * and not the length of the event which would hold the header.
2534  */
2535 int ring_buffer_write(struct ring_buffer *buffer,
2536                         unsigned long length,
2537                         void *data)
2538 {
2539         struct ring_buffer_per_cpu *cpu_buffer;
2540         struct ring_buffer_event *event;
2541         void *body;
2542         int ret = -EBUSY;
2543         int cpu;
2544
2545         if (ring_buffer_flags != RB_BUFFERS_ON)
2546                 return -EBUSY;
2547
2548         preempt_disable_notrace();
2549
2550         if (atomic_read(&buffer->record_disabled))
2551                 goto out;
2552
2553         cpu = raw_smp_processor_id();
2554
2555         if (!cpumask_test_cpu(cpu, buffer->cpumask))
2556                 goto out;
2557
2558         cpu_buffer = buffer->buffers[cpu];
2559
2560         if (atomic_read(&cpu_buffer->record_disabled))
2561                 goto out;
2562
2563         if (length > BUF_MAX_DATA_SIZE)
2564                 goto out;
2565
2566         event = rb_reserve_next_event(buffer, cpu_buffer, length);
2567         if (!event)
2568                 goto out;
2569
2570         body = rb_event_data(event);
2571
2572         memcpy(body, data, length);
2573
2574         rb_commit(cpu_buffer, event);
2575
2576         ret = 0;
2577  out:
2578         preempt_enable_notrace();
2579
2580         return ret;
2581 }
2582 EXPORT_SYMBOL_GPL(ring_buffer_write);
2583
2584 static int rb_per_cpu_empty(struct ring_buffer_per_cpu *cpu_buffer)
2585 {
2586         struct buffer_page *reader = cpu_buffer->reader_page;
2587         struct buffer_page *head = rb_set_head_page(cpu_buffer);
2588         struct buffer_page *commit = cpu_buffer->commit_page;
2589
2590         /* In case of error, head will be NULL */
2591         if (unlikely(!head))
2592                 return 1;
2593
2594         return reader->read == rb_page_commit(reader) &&
2595                 (commit == reader ||
2596                  (commit == head &&
2597                   head->read == rb_page_commit(commit)));
2598 }
2599
2600 /**
2601  * ring_buffer_record_disable - stop all writes into the buffer
2602  * @buffer: The ring buffer to stop writes to.
2603  *
2604  * This prevents all writes to the buffer. Any attempt to write
2605  * to the buffer after this will fail and return NULL.
2606  *
2607  * The caller should call synchronize_sched() after this.
2608  */
2609 void ring_buffer_record_disable(struct ring_buffer *buffer)
2610 {
2611         atomic_inc(&buffer->record_disabled);
2612 }
2613 EXPORT_SYMBOL_GPL(ring_buffer_record_disable);
2614
2615 /**
2616  * ring_buffer_record_enable - enable writes to the buffer
2617  * @buffer: The ring buffer to enable writes
2618  *
2619  * Note, multiple disables will need the same number of enables
2620  * to truly enable the writing (much like preempt_disable).
2621  */
2622 void ring_buffer_record_enable(struct ring_buffer *buffer)
2623 {
2624         atomic_dec(&buffer->record_disabled);
2625 }
2626 EXPORT_SYMBOL_GPL(ring_buffer_record_enable);
2627
2628 /**
2629  * ring_buffer_record_disable_cpu - stop all writes into the cpu_buffer
2630  * @buffer: The ring buffer to stop writes to.
2631  * @cpu: The CPU buffer to stop
2632  *
2633  * This prevents all writes to the buffer. Any attempt to write
2634  * to the buffer after this will fail and return NULL.
2635  *
2636  * The caller should call synchronize_sched() after this.
2637  */
2638 void ring_buffer_record_disable_cpu(struct ring_buffer *buffer, int cpu)
2639 {
2640         struct ring_buffer_per_cpu *cpu_buffer;
2641
2642         if (!cpumask_test_cpu(cpu, buffer->cpumask))
2643                 return;
2644
2645         cpu_buffer = buffer->buffers[cpu];
2646         atomic_inc(&cpu_buffer->record_disabled);
2647 }
2648 EXPORT_SYMBOL_GPL(ring_buffer_record_disable_cpu);
2649
2650 /**
2651  * ring_buffer_record_enable_cpu - enable writes to the buffer
2652  * @buffer: The ring buffer to enable writes
2653  * @cpu: The CPU to enable.
2654  *
2655  * Note, multiple disables will need the same number of enables
2656  * to truly enable the writing (much like preempt_disable).
2657  */
2658 void ring_buffer_record_enable_cpu(struct ring_buffer *buffer, int cpu)
2659 {
2660         struct ring_buffer_per_cpu *cpu_buffer;
2661
2662         if (!cpumask_test_cpu(cpu, buffer->cpumask))
2663                 return;
2664
2665         cpu_buffer = buffer->buffers[cpu];
2666         atomic_dec(&cpu_buffer->record_disabled);
2667 }
2668 EXPORT_SYMBOL_GPL(ring_buffer_record_enable_cpu);
2669
2670 /*
2671  * The total entries in the ring buffer is the running counter
2672  * of entries entered into the ring buffer, minus the sum of
2673  * the entries read from the ring buffer and the number of
2674  * entries that were overwritten.
2675  */
2676 static inline unsigned long
2677 rb_num_of_entries(struct ring_buffer_per_cpu *cpu_buffer)
2678 {
2679         return local_read(&cpu_buffer->entries) -
2680                 (local_read(&cpu_buffer->overrun) + cpu_buffer->read);
2681 }
2682
2683 /**
2684  * ring_buffer_oldest_event_ts - get the oldest event timestamp from the buffer
2685  * @buffer: The ring buffer
2686  * @cpu: The per CPU buffer to read from.
2687  */
2688 unsigned long ring_buffer_oldest_event_ts(struct ring_buffer *buffer, int cpu)
2689 {
2690         unsigned long flags;
2691         struct ring_buffer_per_cpu *cpu_buffer;
2692         struct buffer_page *bpage;
2693         unsigned long ret = 0;
2694
2695         if (!cpumask_test_cpu(cpu, buffer->cpumask))
2696                 return 0;
2697
2698         cpu_buffer = buffer->buffers[cpu];
2699         raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
2700         /*
2701          * if the tail is on reader_page, oldest time stamp is on the reader
2702          * page
2703          */
2704         if (cpu_buffer->tail_page == cpu_buffer->reader_page)
2705                 bpage = cpu_buffer->reader_page;
2706         else
2707                 bpage = rb_set_head_page(cpu_buffer);
2708         if (bpage)
2709                 ret = bpage->page->time_stamp;
2710         raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
2711
2712         return ret;
2713 }
2714 EXPORT_SYMBOL_GPL(ring_buffer_oldest_event_ts);
2715
2716 /**
2717  * ring_buffer_bytes_cpu - get the number of bytes consumed in a cpu buffer
2718  * @buffer: The ring buffer
2719  * @cpu: The per CPU buffer to read from.
2720  */
2721 unsigned long ring_buffer_bytes_cpu(struct ring_buffer *buffer, int cpu)
2722 {
2723         struct ring_buffer_per_cpu *cpu_buffer;
2724         unsigned long ret;
2725
2726         if (!cpumask_test_cpu(cpu, buffer->cpumask))
2727                 return 0;
2728
2729         cpu_buffer = buffer->buffers[cpu];
2730         ret = local_read(&cpu_buffer->entries_bytes) - cpu_buffer->read_bytes;
2731
2732         return ret;
2733 }
2734 EXPORT_SYMBOL_GPL(ring_buffer_bytes_cpu);
2735
2736 /**
2737  * ring_buffer_entries_cpu - get the number of entries in a cpu buffer
2738  * @buffer: The ring buffer
2739  * @cpu: The per CPU buffer to get the entries from.
2740  */
2741 unsigned long ring_buffer_entries_cpu(struct ring_buffer *buffer, int cpu)
2742 {
2743         struct ring_buffer_per_cpu *cpu_buffer;
2744
2745         if (!cpumask_test_cpu(cpu, buffer->cpumask))
2746                 return 0;
2747
2748         cpu_buffer = buffer->buffers[cpu];
2749
2750         return rb_num_of_entries(cpu_buffer);
2751 }
2752 EXPORT_SYMBOL_GPL(ring_buffer_entries_cpu);
2753
2754 /**
2755  * ring_buffer_overrun_cpu - get the number of overruns in a cpu_buffer
2756  * @buffer: The ring buffer
2757  * @cpu: The per CPU buffer to get the number of overruns from
2758  */
2759 unsigned long ring_buffer_overrun_cpu(struct ring_buffer *buffer, int cpu)
2760 {
2761         struct ring_buffer_per_cpu *cpu_buffer;
2762         unsigned long ret;
2763
2764         if (!cpumask_test_cpu(cpu, buffer->cpumask))
2765                 return 0;
2766
2767         cpu_buffer = buffer->buffers[cpu];
2768         ret = local_read(&cpu_buffer->overrun);
2769
2770         return ret;
2771 }
2772 EXPORT_SYMBOL_GPL(ring_buffer_overrun_cpu);
2773
2774 /**
2775  * ring_buffer_commit_overrun_cpu - get the number of overruns caused by commits
2776  * @buffer: The ring buffer
2777  * @cpu: The per CPU buffer to get the number of overruns from
2778  */
2779 unsigned long
2780 ring_buffer_commit_overrun_cpu(struct ring_buffer *buffer, int cpu)
2781 {
2782         struct ring_buffer_per_cpu *cpu_buffer;
2783         unsigned long ret;
2784
2785         if (!cpumask_test_cpu(cpu, buffer->cpumask))
2786                 return 0;
2787
2788         cpu_buffer = buffer->buffers[cpu];
2789         ret = local_read(&cpu_buffer->commit_overrun);
2790
2791         return ret;
2792 }
2793 EXPORT_SYMBOL_GPL(ring_buffer_commit_overrun_cpu);
2794
2795 /**
2796  * ring_buffer_entries - get the number of entries in a buffer
2797  * @buffer: The ring buffer
2798  *
2799  * Returns the total number of entries in the ring buffer
2800  * (all CPU entries)
2801  */
2802 unsigned long ring_buffer_entries(struct ring_buffer *buffer)
2803 {
2804         struct ring_buffer_per_cpu *cpu_buffer;
2805         unsigned long entries = 0;
2806         int cpu;
2807
2808         /* if you care about this being correct, lock the buffer */
2809         for_each_buffer_cpu(buffer, cpu) {
2810                 cpu_buffer = buffer->buffers[cpu];
2811                 entries += rb_num_of_entries(cpu_buffer);
2812         }
2813
2814         return entries;
2815 }
2816 EXPORT_SYMBOL_GPL(ring_buffer_entries);
2817
2818 /**
2819  * ring_buffer_overruns - get the number of overruns in buffer
2820  * @buffer: The ring buffer
2821  *
2822  * Returns the total number of overruns in the ring buffer
2823  * (all CPU entries)
2824  */
2825 unsigned long ring_buffer_overruns(struct ring_buffer *buffer)
2826 {
2827         struct ring_buffer_per_cpu *cpu_buffer;
2828         unsigned long overruns = 0;
2829         int cpu;
2830
2831         /* if you care about this being correct, lock the buffer */
2832         for_each_buffer_cpu(buffer, cpu) {
2833                 cpu_buffer = buffer->buffers[cpu];
2834                 overruns += local_read(&cpu_buffer->overrun);
2835         }
2836
2837         return overruns;
2838 }
2839 EXPORT_SYMBOL_GPL(ring_buffer_overruns);
2840
2841 static void rb_iter_reset(struct ring_buffer_iter *iter)
2842 {
2843         struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
2844
2845         /* Iterator usage is expected to have record disabled */
2846         if (list_empty(&cpu_buffer->reader_page->list)) {
2847                 iter->head_page = rb_set_head_page(cpu_buffer);
2848                 if (unlikely(!iter->head_page))
2849                         return;
2850                 iter->head = iter->head_page->read;
2851         } else {
2852                 iter->head_page = cpu_buffer->reader_page;
2853                 iter->head = cpu_buffer->reader_page->read;
2854         }
2855         if (iter->head)
2856                 iter->read_stamp = cpu_buffer->read_stamp;
2857         else
2858                 iter->read_stamp = iter->head_page->page->time_stamp;
2859         iter->cache_reader_page = cpu_buffer->reader_page;
2860         iter->cache_read = cpu_buffer->read;
2861 }
2862
2863 /**
2864  * ring_buffer_iter_reset - reset an iterator
2865  * @iter: The iterator to reset
2866  *
2867  * Resets the iterator, so that it will start from the beginning
2868  * again.
2869  */
2870 void ring_buffer_iter_reset(struct ring_buffer_iter *iter)
2871 {
2872         struct ring_buffer_per_cpu *cpu_buffer;
2873         unsigned long flags;
2874
2875         if (!iter)
2876                 return;
2877
2878         cpu_buffer = iter->cpu_buffer;
2879
2880         raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
2881         rb_iter_reset(iter);
2882         raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
2883 }
2884 EXPORT_SYMBOL_GPL(ring_buffer_iter_reset);
2885
2886 /**
2887  * ring_buffer_iter_empty - check if an iterator has no more to read
2888  * @iter: The iterator to check
2889  */
2890 int ring_buffer_iter_empty(struct ring_buffer_iter *iter)
2891 {
2892         struct ring_buffer_per_cpu *cpu_buffer;
2893
2894         cpu_buffer = iter->cpu_buffer;
2895
2896         return iter->head_page == cpu_buffer->commit_page &&
2897                 iter->head == rb_commit_index(cpu_buffer);
2898 }
2899 EXPORT_SYMBOL_GPL(ring_buffer_iter_empty);
2900
2901 static void
2902 rb_update_read_stamp(struct ring_buffer_per_cpu *cpu_buffer,
2903                      struct ring_buffer_event *event)
2904 {
2905         u64 delta;
2906
2907         switch (event->type_len) {
2908         case RINGBUF_TYPE_PADDING:
2909                 return;
2910
2911         case RINGBUF_TYPE_TIME_EXTEND:
2912                 delta = event->array[0];
2913                 delta <<= TS_SHIFT;
2914                 delta += event->time_delta;
2915                 cpu_buffer->read_stamp += delta;
2916                 return;
2917
2918         case RINGBUF_TYPE_TIME_STAMP:
2919                 /* FIXME: not implemented */
2920                 return;
2921
2922         case RINGBUF_TYPE_DATA:
2923                 cpu_buffer->read_stamp += event->time_delta;
2924                 return;
2925
2926         default:
2927                 BUG();
2928         }
2929         return;
2930 }
2931
2932 static void
2933 rb_update_iter_read_stamp(struct ring_buffer_iter *iter,
2934                           struct ring_buffer_event *event)
2935 {
2936         u64 delta;
2937
2938         switch (event->type_len) {
2939         case RINGBUF_TYPE_PADDING:
2940                 return;
2941
2942         case RINGBUF_TYPE_TIME_EXTEND:
2943                 delta = event->array[0];
2944                 delta <<= TS_SHIFT;
2945                 delta += event->time_delta;
2946                 iter->read_stamp += delta;
2947                 return;
2948
2949         case RINGBUF_TYPE_TIME_STAMP:
2950                 /* FIXME: not implemented */
2951                 return;
2952
2953         case RINGBUF_TYPE_DATA:
2954                 iter->read_stamp += event->time_delta;
2955                 return;
2956
2957         default:
2958                 BUG();
2959         }
2960         return;
2961 }
2962
2963 static struct buffer_page *
2964 rb_get_reader_page(struct ring_buffer_per_cpu *cpu_buffer)
2965 {
2966         struct buffer_page *reader = NULL;
2967         unsigned long overwrite;
2968         unsigned long flags;
2969         int nr_loops = 0;
2970         int ret;
2971
2972         local_irq_save(flags);
2973         arch_spin_lock(&cpu_buffer->lock);
2974
2975  again:
2976         /*
2977          * This should normally only loop twice. But because the
2978          * start of the reader inserts an empty page, it causes
2979          * a case where we will loop three times. There should be no
2980          * reason to loop four times (that I know of).
2981          */
2982         if (RB_WARN_ON(cpu_buffer, ++nr_loops > 3)) {
2983                 reader = NULL;
2984                 goto out;
2985         }
2986
2987         reader = cpu_buffer->reader_page;
2988
2989         /* If there's more to read, return this page */
2990         if (cpu_buffer->reader_page->read < rb_page_size(reader))
2991                 goto out;
2992
2993         /* Never should we have an index greater than the size */
2994         if (RB_WARN_ON(cpu_buffer,
2995                        cpu_buffer->reader_page->read > rb_page_size(reader)))
2996                 goto out;
2997
2998         /* check if we caught up to the tail */
2999         reader = NULL;
3000         if (cpu_buffer->commit_page == cpu_buffer->reader_page)
3001                 goto out;
3002
3003         /*
3004          * Reset the reader page to size zero.
3005          */
3006         local_set(&cpu_buffer->reader_page->write, 0);
3007         local_set(&cpu_buffer->reader_page->entries, 0);
3008         local_set(&cpu_buffer->reader_page->page->commit, 0);
3009         cpu_buffer->reader_page->real_end = 0;
3010
3011  spin:
3012         /*
3013          * Splice the empty reader page into the list around the head.
3014          */
3015         reader = rb_set_head_page(cpu_buffer);
3016         if (!reader)
3017                 goto out;
3018         cpu_buffer->reader_page->list.next = rb_list_head(reader->list.next);
3019         cpu_buffer->reader_page->list.prev = reader->list.prev;
3020
3021         /*
3022          * cpu_buffer->pages just needs to point to the buffer, it
3023          *  has no specific buffer page to point to. Lets move it out
3024          *  of our way so we don't accidentally swap it.
3025          */
3026         cpu_buffer->pages = reader->list.prev;
3027
3028         /* The reader page will be pointing to the new head */
3029         rb_set_list_to_head(cpu_buffer, &cpu_buffer->reader_page->list);
3030
3031         /*
3032          * We want to make sure we read the overruns after we set up our
3033          * pointers to the next object. The writer side does a
3034          * cmpxchg to cross pages which acts as the mb on the writer
3035          * side. Note, the reader will constantly fail the swap
3036          * while the writer is updating the pointers, so this
3037          * guarantees that the overwrite recorded here is the one we
3038          * want to compare with the last_overrun.
3039          */
3040         smp_mb();
3041         overwrite = local_read(&(cpu_buffer->overrun));
3042
3043         /*
3044          * Here's the tricky part.
3045          *
3046          * We need to move the pointer past the header page.
3047          * But we can only do that if a writer is not currently
3048          * moving it. The page before the header page has the
3049          * flag bit '1' set if it is pointing to the page we want.
3050          * but if the writer is in the process of moving it
3051          * than it will be '2' or already moved '0'.
3052          */
3053
3054         ret = rb_head_page_replace(reader, cpu_buffer->reader_page);
3055
3056         /*
3057          * If we did not convert it, then we must try again.
3058          */
3059         if (!ret)
3060                 goto spin;
3061
3062         /*
3063          * Yeah! We succeeded in replacing the page.
3064          *
3065          * Now make the new head point back to the reader page.
3066          */
3067         rb_list_head(reader->list.next)->prev = &cpu_buffer->reader_page->list;
3068         rb_inc_page(cpu_buffer, &cpu_buffer->head_page);
3069
3070         /* Finally update the reader page to the new head */
3071         cpu_buffer->reader_page = reader;
3072         rb_reset_reader_page(cpu_buffer);
3073
3074         if (overwrite != cpu_buffer->last_overrun) {
3075                 cpu_buffer->lost_events = overwrite - cpu_buffer->last_overrun;
3076                 cpu_buffer->last_overrun = overwrite;
3077         }
3078
3079         goto again;
3080
3081  out:
3082         arch_spin_unlock(&cpu_buffer->lock);
3083         local_irq_restore(flags);
3084
3085         return reader;
3086 }
3087
3088 static void rb_advance_reader(struct ring_buffer_per_cpu *cpu_buffer)
3089 {
3090         struct ring_buffer_event *event;
3091         struct buffer_page *reader;
3092         unsigned length;
3093
3094         reader = rb_get_reader_page(cpu_buffer);
3095
3096         /* This function should not be called when buffer is empty */
3097         if (RB_WARN_ON(cpu_buffer, !reader))
3098                 return;
3099
3100         event = rb_reader_event(cpu_buffer);
3101
3102         if (event->type_len <= RINGBUF_TYPE_DATA_TYPE_LEN_MAX)
3103                 cpu_buffer->read++;
3104
3105         rb_update_read_stamp(cpu_buffer, event);
3106
3107         length = rb_event_length(event);
3108         cpu_buffer->reader_page->read += length;
3109 }
3110
3111 static void rb_advance_iter(struct ring_buffer_iter *iter)
3112 {
3113         struct ring_buffer_per_cpu *cpu_buffer;
3114         struct ring_buffer_event *event;
3115         unsigned length;
3116
3117         cpu_buffer = iter->cpu_buffer;
3118
3119         /*
3120          * Check if we are at the end of the buffer.
3121          */
3122         if (iter->head >= rb_page_size(iter->head_page)) {
3123                 /* discarded commits can make the page empty */
3124                 if (iter->head_page == cpu_buffer->commit_page)
3125                         return;
3126                 rb_inc_iter(iter);
3127                 return;
3128         }
3129
3130         event = rb_iter_head_event(iter);
3131
3132         length = rb_event_length(event);
3133
3134         /*
3135          * This should not be called to advance the header if we are
3136          * at the tail of the buffer.
3137          */
3138         if (RB_WARN_ON(cpu_buffer,
3139                        (iter->head_page == cpu_buffer->commit_page) &&
3140                        (iter->head + length > rb_commit_index(cpu_buffer))))
3141                 return;
3142
3143         rb_update_iter_read_stamp(iter, event);
3144
3145         iter->head += length;
3146
3147         /* check for end of page padding */
3148         if ((iter->head >= rb_page_size(iter->head_page)) &&
3149             (iter->head_page != cpu_buffer->commit_page))
3150                 rb_advance_iter(iter);
3151 }
3152
3153 static int rb_lost_events(struct ring_buffer_per_cpu *cpu_buffer)
3154 {
3155         return cpu_buffer->lost_events;
3156 }
3157
3158 static struct ring_buffer_event *
3159 rb_buffer_peek(struct ring_buffer_per_cpu *cpu_buffer, u64 *ts,
3160                unsigned long *lost_events)
3161 {
3162         struct ring_buffer_event *event;
3163         struct buffer_page *reader;
3164         int nr_loops = 0;
3165
3166  again:
3167         /*
3168          * We repeat when a time extend is encountered.
3169          * Since the time extend is always attached to a data event,
3170          * we should never loop more than once.
3171          * (We never hit the following condition more than twice).
3172          */
3173         if (RB_WARN_ON(cpu_buffer, ++nr_loops > 2))
3174                 return NULL;
3175
3176         reader = rb_get_reader_page(cpu_buffer);
3177         if (!reader)
3178                 return NULL;
3179
3180         event = rb_reader_event(cpu_buffer);
3181
3182         switch (event->type_len) {
3183         case RINGBUF_TYPE_PADDING:
3184                 if (rb_null_event(event))
3185                         RB_WARN_ON(cpu_buffer, 1);
3186                 /*
3187                  * Because the writer could be discarding every
3188                  * event it creates (which would probably be bad)
3189                  * if we were to go back to "again" then we may never
3190                  * catch up, and will trigger the warn on, or lock
3191                  * the box. Return the padding, and we will release
3192                  * the current locks, and try again.
3193                  */
3194                 return event;
3195
3196         case RINGBUF_TYPE_TIME_EXTEND:
3197                 /* Internal data, OK to advance */
3198                 rb_advance_reader(cpu_buffer);
3199                 goto again;
3200
3201         case RINGBUF_TYPE_TIME_STAMP:
3202                 /* FIXME: not implemented */
3203                 rb_advance_reader(cpu_buffer);
3204                 goto again;
3205
3206         case RINGBUF_TYPE_DATA:
3207                 if (ts) {
3208                         *ts = cpu_buffer->read_stamp + event->time_delta;
3209                         ring_buffer_normalize_time_stamp(cpu_buffer->buffer,
3210                                                          cpu_buffer->cpu, ts);
3211                 }
3212                 if (lost_events)
3213                         *lost_events = rb_lost_events(cpu_buffer);
3214                 return event;
3215
3216         default:
3217                 BUG();
3218         }
3219
3220         return NULL;
3221 }
3222 EXPORT_SYMBOL_GPL(ring_buffer_peek);
3223
3224 static struct ring_buffer_event *
3225 rb_iter_peek(struct ring_buffer_iter *iter, u64 *ts)
3226 {
3227         struct ring_buffer *buffer;
3228         struct ring_buffer_per_cpu *cpu_buffer;
3229         struct ring_buffer_event *event;
3230         int nr_loops = 0;
3231
3232         cpu_buffer = iter->cpu_buffer;
3233         buffer = cpu_buffer->buffer;
3234
3235         /*
3236          * Check if someone performed a consuming read to
3237          * the buffer. A consuming read invalidates the iterator
3238          * and we need to reset the iterator in this case.
3239          */
3240         if (unlikely(iter->cache_read != cpu_buffer->read ||
3241                      iter->cache_reader_page != cpu_buffer->reader_page))
3242                 rb_iter_reset(iter);
3243
3244  again:
3245         if (ring_buffer_iter_empty(iter))
3246                 return NULL;
3247
3248         /*
3249          * We repeat when a time extend is encountered.
3250          * Since the time extend is always attached to a data event,
3251          * we should never loop more than once.
3252          * (We never hit the following condition more than twice).
3253          */
3254         if (RB_WARN_ON(cpu_buffer, ++nr_loops > 2))
3255                 return NULL;
3256
3257         if (rb_per_cpu_empty(cpu_buffer))
3258                 return NULL;
3259
3260         if (iter->head >= local_read(&iter->head_page->page->commit)) {
3261                 rb_inc_iter(iter);
3262                 goto again;
3263         }
3264
3265         event = rb_iter_head_event(iter);
3266
3267         switch (event->type_len) {
3268         case RINGBUF_TYPE_PADDING:
3269                 if (rb_null_event(event)) {
3270                         rb_inc_iter(iter);
3271                         goto again;
3272                 }
3273                 rb_advance_iter(iter);
3274                 return event;
3275
3276         case RINGBUF_TYPE_TIME_EXTEND:
3277                 /* Internal data, OK to advance */
3278                 rb_advance_iter(iter);
3279                 goto again;
3280
3281         case RINGBUF_TYPE_TIME_STAMP:
3282                 /* FIXME: not implemented */
3283                 rb_advance_iter(iter);
3284                 goto again;
3285
3286         case RINGBUF_TYPE_DATA:
3287                 if (ts) {
3288                         *ts = iter->read_stamp + event->time_delta;
3289                         ring_buffer_normalize_time_stamp(buffer,
3290                                                          cpu_buffer->cpu, ts);
3291                 }
3292                 return event;
3293
3294         default:
3295                 BUG();
3296         }
3297
3298         return NULL;
3299 }
3300 EXPORT_SYMBOL_GPL(ring_buffer_iter_peek);
3301
3302 static inline int rb_ok_to_lock(void)
3303 {
3304         /*
3305          * If an NMI die dumps out the content of the ring buffer
3306          * do not grab locks. We also permanently disable the ring
3307          * buffer too. A one time deal is all you get from reading
3308          * the ring buffer from an NMI.
3309          */
3310         if (likely(!in_nmi()))
3311                 return 1;
3312
3313         tracing_off_permanent();
3314         return 0;
3315 }
3316
3317 /**
3318  * ring_buffer_peek - peek at the next event to be read
3319  * @buffer: The ring buffer to read
3320  * @cpu: The cpu to peak at
3321  * @ts: The timestamp counter of this event.
3322  * @lost_events: a variable to store if events were lost (may be NULL)
3323  *
3324  * This will return the event that will be read next, but does
3325  * not consume the data.
3326  */
3327 struct ring_buffer_event *
3328 ring_buffer_peek(struct ring_buffer *buffer, int cpu, u64 *ts,
3329                  unsigned long *lost_events)
3330 {
3331         struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
3332         struct ring_buffer_event *event;
3333         unsigned long flags;
3334         int dolock;
3335
3336         if (!cpumask_test_cpu(cpu, buffer->cpumask))
3337                 return NULL;
3338
3339         dolock = rb_ok_to_lock();
3340  again:
3341         local_irq_save(flags);
3342         if (dolock)
3343                 raw_spin_lock(&cpu_buffer->reader_lock);
3344         event = rb_buffer_peek(cpu_buffer, ts, lost_events);
3345         if (event && event->type_len == RINGBUF_TYPE_PADDING)
3346                 rb_advance_reader(cpu_buffer);
3347         if (dolock)
3348                 raw_spin_unlock(&cpu_buffer->reader_lock);
3349         local_irq_restore(flags);
3350
3351         if (event && event->type_len == RINGBUF_TYPE_PADDING)
3352                 goto again;
3353
3354         return event;
3355 }
3356
3357 /**
3358  * ring_buffer_iter_peek - peek at the next event to be read
3359  * @iter: The ring buffer iterator
3360  * @ts: The timestamp counter of this event.
3361  *
3362  * This will return the event that will be read next, but does
3363  * not increment the iterator.
3364  */
3365 struct ring_buffer_event *
3366 ring_buffer_iter_peek(struct ring_buffer_iter *iter, u64 *ts)
3367 {
3368         struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
3369         struct ring_buffer_event *event;
3370         unsigned long flags;
3371
3372  again:
3373         raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
3374         event = rb_iter_peek(iter, ts);
3375         raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
3376
3377         if (event && event->type_len == RINGBUF_TYPE_PADDING)
3378                 goto again;
3379
3380         return event;
3381 }
3382
3383 /**
3384  * ring_buffer_consume - return an event and consume it
3385  * @buffer: The ring buffer to get the next event from
3386  * @cpu: the cpu to read the buffer from
3387  * @ts: a variable to store the timestamp (may be NULL)
3388  * @lost_events: a variable to store if events were lost (may be NULL)
3389  *
3390  * Returns the next event in the ring buffer, and that event is consumed.
3391  * Meaning, that sequential reads will keep returning a different event,
3392  * and eventually empty the ring buffer if the producer is slower.
3393  */
3394 struct ring_buffer_event *
3395 ring_buffer_consume(struct ring_buffer *buffer, int cpu, u64 *ts,
3396                     unsigned long *lost_events)
3397 {
3398         struct ring_buffer_per_cpu *cpu_buffer;
3399         struct ring_buffer_event *event = NULL;
3400         unsigned long flags;
3401         int dolock;
3402
3403         dolock = rb_ok_to_lock();
3404
3405  again:
3406         /* might be called in atomic */
3407         preempt_disable();
3408
3409         if (!cpumask_test_cpu(cpu, buffer->cpumask))
3410                 goto out;
3411
3412         cpu_buffer = buffer->buffers[cpu];
3413         local_irq_save(flags);
3414         if (dolock)
3415                 raw_spin_lock(&cpu_buffer->reader_lock);
3416
3417         event = rb_buffer_peek(cpu_buffer, ts, lost_events);
3418         if (event) {
3419                 cpu_buffer->lost_events = 0;
3420                 rb_advance_reader(cpu_buffer);
3421         }
3422
3423         if (dolock)
3424                 raw_spin_unlock(&cpu_buffer->reader_lock);
3425         local_irq_restore(flags);
3426
3427  out:
3428         preempt_enable();
3429
3430         if (event && event->type_len == RINGBUF_TYPE_PADDING)
3431                 goto again;
3432
3433         return event;
3434 }
3435 EXPORT_SYMBOL_GPL(ring_buffer_consume);
3436
3437 /**
3438  * ring_buffer_read_prepare - Prepare for a non consuming read of the buffer
3439  * @buffer: The ring buffer to read from
3440  * @cpu: The cpu buffer to iterate over
3441  *
3442  * This performs the initial preparations necessary to iterate
3443  * through the buffer.  Memory is allocated, buffer recording
3444  * is disabled, and the iterator pointer is returned to the caller.
3445  *
3446  * Disabling buffer recordng prevents the reading from being
3447  * corrupted. This is not a consuming read, so a producer is not
3448  * expected.
3449  *
3450  * After a sequence of ring_buffer_read_prepare calls, the user is
3451  * expected to make at least one call to ring_buffer_prepare_sync.
3452  * Afterwards, ring_buffer_read_start is invoked to get things going
3453  * for real.
3454  *
3455  * This overall must be paired with ring_buffer_finish.
3456  */
3457 struct ring_buffer_iter *
3458 ring_buffer_read_prepare(struct ring_buffer *buffer, int cpu)
3459 {
3460         struct ring_buffer_per_cpu *cpu_buffer;
3461         struct ring_buffer_iter *iter;
3462
3463         if (!cpumask_test_cpu(cpu, buffer->cpumask))
3464                 return NULL;
3465
3466         iter = kmalloc(sizeof(*iter), GFP_KERNEL);
3467         if (!iter)
3468                 return NULL;
3469
3470         cpu_buffer = buffer->buffers[cpu];
3471
3472         iter->cpu_buffer = cpu_buffer;
3473
3474         atomic_inc(&cpu_buffer->record_disabled);
3475
3476         return iter;
3477 }
3478 EXPORT_SYMBOL_GPL(ring_buffer_read_prepare);
3479
3480 /**
3481  * ring_buffer_read_prepare_sync - Synchronize a set of prepare calls
3482  *
3483  * All previously invoked ring_buffer_read_prepare calls to prepare
3484  * iterators will be synchronized.  Afterwards, read_buffer_read_start
3485  * calls on those iterators are allowed.
3486  */
3487 void
3488 ring_buffer_read_prepare_sync(void)
3489 {
3490         synchronize_sched();
3491 }
3492 EXPORT_SYMBOL_GPL(ring_buffer_read_prepare_sync);
3493
3494 /**
3495  * ring_buffer_read_start - start a non consuming read of the buffer
3496  * @iter: The iterator returned by ring_buffer_read_prepare
3497  *
3498  * This finalizes the startup of an iteration through the buffer.
3499  * The iterator comes from a call to ring_buffer_read_prepare and
3500  * an intervening ring_buffer_read_prepare_sync must have been
3501  * performed.
3502  *
3503  * Must be paired with ring_buffer_finish.
3504  */
3505 void
3506 ring_buffer_read_start(struct ring_buffer_iter *iter)
3507 {
3508         struct ring_buffer_per_cpu *cpu_buffer;
3509         unsigned long flags;
3510
3511         if (!iter)
3512                 return;
3513
3514         cpu_buffer = iter->cpu_buffer;
3515
3516         raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
3517         arch_spin_lock(&cpu_buffer->lock);
3518         rb_iter_reset(iter);
3519         arch_spin_unlock(&cpu_buffer->lock);
3520         raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
3521 }
3522 EXPORT_SYMBOL_GPL(ring_buffer_read_start);
3523
3524 /**
3525  * ring_buffer_finish - finish reading the iterator of the buffer
3526  * @iter: The iterator retrieved by ring_buffer_start
3527  *
3528  * This re-enables the recording to the buffer, and frees the
3529  * iterator.
3530  */
3531 void
3532 ring_buffer_read_finish(struct ring_buffer_iter *iter)
3533 {
3534         struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
3535
3536         atomic_dec(&cpu_buffer->record_disabled);
3537         kfree(iter);
3538 }
3539 EXPORT_SYMBOL_GPL(ring_buffer_read_finish);
3540
3541 /**
3542  * ring_buffer_read - read the next item in the ring buffer by the iterator
3543  * @iter: The ring buffer iterator
3544  * @ts: The time stamp of the event read.
3545  *
3546  * This reads the next event in the ring buffer and increments the iterator.
3547  */
3548 struct ring_buffer_event *
3549 ring_buffer_read(struct ring_buffer_iter *iter, u64 *ts)
3550 {
3551         struct ring_buffer_event *event;
3552         struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
3553         unsigned long flags;
3554
3555         raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
3556  again:
3557         event = rb_iter_peek(iter, ts);
3558         if (!event)
3559                 goto out;
3560
3561         if (event->type_len == RINGBUF_TYPE_PADDING)
3562                 goto again;
3563
3564         rb_advance_iter(iter);
3565  out:
3566         raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
3567
3568         return event;
3569 }
3570 EXPORT_SYMBOL_GPL(ring_buffer_read);
3571
3572 /**
3573  * ring_buffer_size - return the size of the ring buffer (in bytes)
3574  * @buffer: The ring buffer.
3575  */
3576 unsigned long ring_buffer_size(struct ring_buffer *buffer)
3577 {
3578         return BUF_PAGE_SIZE * buffer->pages;
3579 }
3580 EXPORT_SYMBOL_GPL(ring_buffer_size);
3581
3582 static void
3583 rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer)
3584 {
3585         rb_head_page_deactivate(cpu_buffer);
3586
3587         cpu_buffer->head_page
3588                 = list_entry(cpu_buffer->pages, struct buffer_page, list);
3589         local_set(&cpu_buffer->head_page->write, 0);
3590         local_set(&cpu_buffer->head_page->entries, 0);
3591         local_set(&cpu_buffer->head_page->page->commit, 0);
3592
3593         cpu_buffer->head_page->read = 0;
3594
3595         cpu_buffer->tail_page = cpu_buffer->head_page;
3596         cpu_buffer->commit_page = cpu_buffer->head_page;
3597
3598         INIT_LIST_HEAD(&cpu_buffer->reader_page->list);
3599         local_set(&cpu_buffer->reader_page->write, 0);
3600         local_set(&cpu_buffer->reader_page->entries, 0);
3601         local_set(&cpu_buffer->reader_page->page->commit, 0);
3602         cpu_buffer->reader_page->read = 0;
3603
3604         local_set(&cpu_buffer->commit_overrun, 0);
3605         local_set(&cpu_buffer->entries_bytes, 0);
3606         local_set(&cpu_buffer->overrun, 0);
3607         local_set(&cpu_buffer->entries, 0);
3608         local_set(&cpu_buffer->committing, 0);
3609         local_set(&cpu_buffer->commits, 0);
3610         cpu_buffer->read = 0;
3611         cpu_buffer->read_bytes = 0;
3612
3613         cpu_buffer->write_stamp = 0;
3614         cpu_buffer->read_stamp = 0;
3615
3616         cpu_buffer->lost_events = 0;
3617         cpu_buffer->last_overrun = 0;
3618
3619         rb_head_page_activate(cpu_buffer);
3620 }
3621
3622 /**
3623  * ring_buffer_reset_cpu - reset a ring buffer per CPU buffer
3624  * @buffer: The ring buffer to reset a per cpu buffer of
3625  * @cpu: The CPU buffer to be reset
3626  */
3627 void ring_buffer_reset_cpu(struct ring_buffer *buffer, int cpu)
3628 {
3629         struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
3630         unsigned long flags;
3631
3632         if (!cpumask_test_cpu(cpu, buffer->cpumask))
3633                 return;
3634
3635         atomic_inc(&cpu_buffer->record_disabled);
3636
3637         raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
3638
3639         if (RB_WARN_ON(cpu_buffer, local_read(&cpu_buffer->committing)))
3640                 goto out;
3641
3642         arch_spin_lock(&cpu_buffer->lock);
3643
3644         rb_reset_cpu(cpu_buffer);
3645
3646         arch_spin_unlock(&cpu_buffer->lock);
3647
3648  out:
3649         raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
3650
3651         atomic_dec(&cpu_buffer->record_disabled);
3652 }
3653 EXPORT_SYMBOL_GPL(ring_buffer_reset_cpu);
3654
3655 /**
3656  * ring_buffer_reset - reset a ring buffer
3657  * @buffer: The ring buffer to reset all cpu buffers
3658  */
3659 void ring_buffer_reset(struct ring_buffer *buffer)
3660 {
3661         int cpu;
3662
3663         for_each_buffer_cpu(buffer, cpu)
3664                 ring_buffer_reset_cpu(buffer, cpu);
3665 }
3666 EXPORT_SYMBOL_GPL(ring_buffer_reset);
3667
3668 /**
3669  * rind_buffer_empty - is the ring buffer empty?
3670  * @buffer: The ring buffer to test
3671  */
3672 int ring_buffer_empty(struct ring_buffer *buffer)
3673 {
3674         struct ring_buffer_per_cpu *cpu_buffer;
3675         unsigned long flags;
3676         int dolock;
3677         int cpu;
3678         int ret;
3679
3680         dolock = rb_ok_to_lock();
3681
3682         /* yes this is racy, but if you don't like the race, lock the buffer */
3683         for_each_buffer_cpu(buffer, cpu) {
3684                 cpu_buffer = buffer->buffers[cpu];
3685                 local_irq_save(flags);
3686                 if (dolock)
3687                         raw_spin_lock(&cpu_buffer->reader_lock);
3688                 ret = rb_per_cpu_empty(cpu_buffer);
3689                 if (dolock)
3690                         raw_spin_unlock(&cpu_buffer->reader_lock);
3691                 local_irq_restore(flags);
3692
3693                 if (!ret)
3694                         return 0;
3695         }
3696
3697         return 1;
3698 }
3699 EXPORT_SYMBOL_GPL(ring_buffer_empty);
3700
3701 /**
3702  * ring_buffer_empty_cpu - is a cpu buffer of a ring buffer empty?
3703  * @buffer: The ring buffer
3704  * @cpu: The CPU buffer to test
3705  */
3706 int ring_buffer_empty_cpu(struct ring_buffer *buffer, int cpu)
3707 {
3708         struct ring_buffer_per_cpu *cpu_buffer;
3709         unsigned long flags;
3710         int dolock;
3711         int ret;
3712
3713         if (!cpumask_test_cpu(cpu, buffer->cpumask))
3714                 return 1;
3715
3716         dolock = rb_ok_to_lock();
3717
3718         cpu_buffer = buffer->buffers[cpu];
3719         local_irq_save(flags);
3720         if (dolock)
3721                 raw_spin_lock(&cpu_buffer->reader_lock);
3722         ret = rb_per_cpu_empty(cpu_buffer);
3723         if (dolock)
3724                 raw_spin_unlock(&cpu_buffer->reader_lock);
3725         local_irq_restore(flags);
3726
3727         return ret;
3728 }
3729 EXPORT_SYMBOL_GPL(ring_buffer_empty_cpu);
3730
3731 #ifdef CONFIG_RING_BUFFER_ALLOW_SWAP
3732 /**
3733  * ring_buffer_swap_cpu - swap a CPU buffer between two ring buffers
3734  * @buffer_a: One buffer to swap with
3735  * @buffer_b: The other buffer to swap with
3736  *
3737  * This function is useful for tracers that want to take a "snapshot"
3738  * of a CPU buffer and has another back up buffer lying around.
3739  * it is expected that the tracer handles the cpu buffer not being
3740  * used at the moment.
3741  */
3742 int ring_buffer_swap_cpu(struct ring_buffer *buffer_a,
3743                          struct ring_buffer *buffer_b, int cpu)
3744 {
3745         struct ring_buffer_per_cpu *cpu_buffer_a;
3746         struct ring_buffer_per_cpu *cpu_buffer_b;
3747         int ret = -EINVAL;
3748
3749         if (!cpumask_test_cpu(cpu, buffer_a->cpumask) ||
3750             !cpumask_test_cpu(cpu, buffer_b->cpumask))
3751                 goto out;
3752
3753         /* At least make sure the two buffers are somewhat the same */
3754         if (buffer_a->pages != buffer_b->pages)
3755                 goto out;
3756
3757         ret = -EAGAIN;
3758
3759         if (ring_buffer_flags != RB_BUFFERS_ON)
3760                 goto out;
3761
3762         if (atomic_read(&buffer_a->record_disabled))
3763                 goto out;
3764
3765         if (atomic_read(&buffer_b->record_disabled))
3766                 goto out;
3767
3768         cpu_buffer_a = buffer_a->buffers[cpu];
3769         cpu_buffer_b = buffer_b->buffers[cpu];
3770
3771         if (atomic_read(&cpu_buffer_a->record_disabled))
3772                 goto out;
3773
3774         if (atomic_read(&cpu_buffer_b->record_disabled))
3775                 goto out;
3776
3777         /*
3778          * We can't do a synchronize_sched here because this
3779          * function can be called in atomic context.
3780          * Normally this will be called from the same CPU as cpu.
3781          * If not it's up to the caller to protect this.
3782          */
3783         atomic_inc(&cpu_buffer_a->record_disabled);
3784         atomic_inc(&cpu_buffer_b->record_disabled);
3785
3786         ret = -EBUSY;
3787         if (local_read(&cpu_buffer_a->committing))
3788                 goto out_dec;
3789         if (local_read(&cpu_buffer_b->committing))
3790                 goto out_dec;
3791
3792         buffer_a->buffers[cpu] = cpu_buffer_b;
3793         buffer_b->buffers[cpu] = cpu_buffer_a;
3794
3795         cpu_buffer_b->buffer = buffer_a;
3796         cpu_buffer_a->buffer = buffer_b;
3797
3798         ret = 0;
3799
3800 out_dec:
3801         atomic_dec(&cpu_buffer_a->record_disabled);
3802         atomic_dec(&cpu_buffer_b->record_disabled);
3803 out:
3804         return ret;
3805 }
3806 EXPORT_SYMBOL_GPL(ring_buffer_swap_cpu);
3807 #endif /* CONFIG_RING_BUFFER_ALLOW_SWAP */
3808
3809 /**
3810  * ring_buffer_alloc_read_page - allocate a page to read from buffer
3811  * @buffer: the buffer to allocate for.
3812  *
3813  * This function is used in conjunction with ring_buffer_read_page.
3814  * When reading a full page from the ring buffer, these functions