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