d1187ef20caf914048fb897dc5c63f1b302edbf6
[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                         return;
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                 return;
1220
1221         rb_reset_cpu(cpu_buffer);
1222         rb_check_pages(cpu_buffer);
1223
1224         spin_unlock_irq(&cpu_buffer->reader_lock);
1225 }
1226
1227 static void
1228 rb_insert_pages(struct ring_buffer_per_cpu *cpu_buffer,
1229                 struct list_head *pages, unsigned nr_pages)
1230 {
1231         struct buffer_page *bpage;
1232         struct list_head *p;
1233         unsigned i;
1234
1235         spin_lock_irq(&cpu_buffer->reader_lock);
1236         rb_head_page_deactivate(cpu_buffer);
1237
1238         for (i = 0; i < nr_pages; i++) {
1239                 if (RB_WARN_ON(cpu_buffer, list_empty(pages)))
1240                         return;
1241                 p = pages->next;
1242                 bpage = list_entry(p, struct buffer_page, list);
1243                 list_del_init(&bpage->list);
1244                 list_add_tail(&bpage->list, cpu_buffer->pages);
1245         }
1246         rb_reset_cpu(cpu_buffer);
1247         rb_check_pages(cpu_buffer);
1248
1249         spin_unlock_irq(&cpu_buffer->reader_lock);
1250 }
1251
1252 /**
1253  * ring_buffer_resize - resize the ring buffer
1254  * @buffer: the buffer to resize.
1255  * @size: the new size.
1256  *
1257  * Minimum size is 2 * BUF_PAGE_SIZE.
1258  *
1259  * Returns -1 on failure.
1260  */
1261 int ring_buffer_resize(struct ring_buffer *buffer, unsigned long size)
1262 {
1263         struct ring_buffer_per_cpu *cpu_buffer;
1264         unsigned nr_pages, rm_pages, new_pages;
1265         struct buffer_page *bpage, *tmp;
1266         unsigned long buffer_size;
1267         unsigned long addr;
1268         LIST_HEAD(pages);
1269         int i, cpu;
1270
1271         /*
1272          * Always succeed at resizing a non-existent buffer:
1273          */
1274         if (!buffer)
1275                 return size;
1276
1277         size = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
1278         size *= BUF_PAGE_SIZE;
1279         buffer_size = buffer->pages * BUF_PAGE_SIZE;
1280
1281         /* we need a minimum of two pages */
1282         if (size < BUF_PAGE_SIZE * 2)
1283                 size = BUF_PAGE_SIZE * 2;
1284
1285         if (size == buffer_size)
1286                 return size;
1287
1288         atomic_inc(&buffer->record_disabled);
1289
1290         /* Make sure all writers are done with this buffer. */
1291         synchronize_sched();
1292
1293         mutex_lock(&buffer->mutex);
1294         get_online_cpus();
1295
1296         nr_pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
1297
1298         if (size < buffer_size) {
1299
1300                 /* easy case, just free pages */
1301                 if (RB_WARN_ON(buffer, nr_pages >= buffer->pages))
1302                         goto out_fail;
1303
1304                 rm_pages = buffer->pages - nr_pages;
1305
1306                 for_each_buffer_cpu(buffer, cpu) {
1307                         cpu_buffer = buffer->buffers[cpu];
1308                         rb_remove_pages(cpu_buffer, rm_pages);
1309                 }
1310                 goto out;
1311         }
1312
1313         /*
1314          * This is a bit more difficult. We only want to add pages
1315          * when we can allocate enough for all CPUs. We do this
1316          * by allocating all the pages and storing them on a local
1317          * link list. If we succeed in our allocation, then we
1318          * add these pages to the cpu_buffers. Otherwise we just free
1319          * them all and return -ENOMEM;
1320          */
1321         if (RB_WARN_ON(buffer, nr_pages <= buffer->pages))
1322                 goto out_fail;
1323
1324         new_pages = nr_pages - buffer->pages;
1325
1326         for_each_buffer_cpu(buffer, cpu) {
1327                 for (i = 0; i < new_pages; i++) {
1328                         bpage = kzalloc_node(ALIGN(sizeof(*bpage),
1329                                                   cache_line_size()),
1330                                             GFP_KERNEL, cpu_to_node(cpu));
1331                         if (!bpage)
1332                                 goto free_pages;
1333                         list_add(&bpage->list, &pages);
1334                         addr = __get_free_page(GFP_KERNEL);
1335                         if (!addr)
1336                                 goto free_pages;
1337                         bpage->page = (void *)addr;
1338                         rb_init_page(bpage->page);
1339                 }
1340         }
1341
1342         for_each_buffer_cpu(buffer, cpu) {
1343                 cpu_buffer = buffer->buffers[cpu];
1344                 rb_insert_pages(cpu_buffer, &pages, new_pages);
1345         }
1346
1347         if (RB_WARN_ON(buffer, !list_empty(&pages)))
1348                 goto out_fail;
1349
1350  out:
1351         buffer->pages = nr_pages;
1352         put_online_cpus();
1353         mutex_unlock(&buffer->mutex);
1354
1355         atomic_dec(&buffer->record_disabled);
1356
1357         return size;
1358
1359  free_pages:
1360         list_for_each_entry_safe(bpage, tmp, &pages, list) {
1361                 list_del_init(&bpage->list);
1362                 free_buffer_page(bpage);
1363         }
1364         put_online_cpus();
1365         mutex_unlock(&buffer->mutex);
1366         atomic_dec(&buffer->record_disabled);
1367         return -ENOMEM;
1368
1369         /*
1370          * Something went totally wrong, and we are too paranoid
1371          * to even clean up the mess.
1372          */
1373  out_fail:
1374         put_online_cpus();
1375         mutex_unlock(&buffer->mutex);
1376         atomic_dec(&buffer->record_disabled);
1377         return -1;
1378 }
1379 EXPORT_SYMBOL_GPL(ring_buffer_resize);
1380
1381 static inline void *
1382 __rb_data_page_index(struct buffer_data_page *bpage, unsigned index)
1383 {
1384         return bpage->data + index;
1385 }
1386
1387 static inline void *__rb_page_index(struct buffer_page *bpage, unsigned index)
1388 {
1389         return bpage->page->data + index;
1390 }
1391
1392 static inline struct ring_buffer_event *
1393 rb_reader_event(struct ring_buffer_per_cpu *cpu_buffer)
1394 {
1395         return __rb_page_index(cpu_buffer->reader_page,
1396                                cpu_buffer->reader_page->read);
1397 }
1398
1399 static inline struct ring_buffer_event *
1400 rb_iter_head_event(struct ring_buffer_iter *iter)
1401 {
1402         return __rb_page_index(iter->head_page, iter->head);
1403 }
1404
1405 static inline unsigned long rb_page_write(struct buffer_page *bpage)
1406 {
1407         return local_read(&bpage->write) & RB_WRITE_MASK;
1408 }
1409
1410 static inline unsigned rb_page_commit(struct buffer_page *bpage)
1411 {
1412         return local_read(&bpage->page->commit);
1413 }
1414
1415 static inline unsigned long rb_page_entries(struct buffer_page *bpage)
1416 {
1417         return local_read(&bpage->entries) & RB_WRITE_MASK;
1418 }
1419
1420 /* Size is determined by what has been commited */
1421 static inline unsigned rb_page_size(struct buffer_page *bpage)
1422 {
1423         return rb_page_commit(bpage);
1424 }
1425
1426 static inline unsigned
1427 rb_commit_index(struct ring_buffer_per_cpu *cpu_buffer)
1428 {
1429         return rb_page_commit(cpu_buffer->commit_page);
1430 }
1431
1432 static inline unsigned
1433 rb_event_index(struct ring_buffer_event *event)
1434 {
1435         unsigned long addr = (unsigned long)event;
1436
1437         return (addr & ~PAGE_MASK) - BUF_PAGE_HDR_SIZE;
1438 }
1439
1440 static inline int
1441 rb_event_is_commit(struct ring_buffer_per_cpu *cpu_buffer,
1442                    struct ring_buffer_event *event)
1443 {
1444         unsigned long addr = (unsigned long)event;
1445         unsigned long index;
1446
1447         index = rb_event_index(event);
1448         addr &= PAGE_MASK;
1449
1450         return cpu_buffer->commit_page->page == (void *)addr &&
1451                 rb_commit_index(cpu_buffer) == index;
1452 }
1453
1454 static void
1455 rb_set_commit_to_write(struct ring_buffer_per_cpu *cpu_buffer)
1456 {
1457         unsigned long max_count;
1458
1459         /*
1460          * We only race with interrupts and NMIs on this CPU.
1461          * If we own the commit event, then we can commit
1462          * all others that interrupted us, since the interruptions
1463          * are in stack format (they finish before they come
1464          * back to us). This allows us to do a simple loop to
1465          * assign the commit to the tail.
1466          */
1467  again:
1468         max_count = cpu_buffer->buffer->pages * 100;
1469
1470         while (cpu_buffer->commit_page != cpu_buffer->tail_page) {
1471                 if (RB_WARN_ON(cpu_buffer, !(--max_count)))
1472                         return;
1473                 if (RB_WARN_ON(cpu_buffer,
1474                                rb_is_reader_page(cpu_buffer->tail_page)))
1475                         return;
1476                 local_set(&cpu_buffer->commit_page->page->commit,
1477                           rb_page_write(cpu_buffer->commit_page));
1478                 rb_inc_page(cpu_buffer, &cpu_buffer->commit_page);
1479                 cpu_buffer->write_stamp =
1480                         cpu_buffer->commit_page->page->time_stamp;
1481                 /* add barrier to keep gcc from optimizing too much */
1482                 barrier();
1483         }
1484         while (rb_commit_index(cpu_buffer) !=
1485                rb_page_write(cpu_buffer->commit_page)) {
1486
1487                 local_set(&cpu_buffer->commit_page->page->commit,
1488                           rb_page_write(cpu_buffer->commit_page));
1489                 RB_WARN_ON(cpu_buffer,
1490                            local_read(&cpu_buffer->commit_page->page->commit) &
1491                            ~RB_WRITE_MASK);
1492                 barrier();
1493         }
1494
1495         /* again, keep gcc from optimizing */
1496         barrier();
1497
1498         /*
1499          * If an interrupt came in just after the first while loop
1500          * and pushed the tail page forward, we will be left with
1501          * a dangling commit that will never go forward.
1502          */
1503         if (unlikely(cpu_buffer->commit_page != cpu_buffer->tail_page))
1504                 goto again;
1505 }
1506
1507 static void rb_reset_reader_page(struct ring_buffer_per_cpu *cpu_buffer)
1508 {
1509         cpu_buffer->read_stamp = cpu_buffer->reader_page->page->time_stamp;
1510         cpu_buffer->reader_page->read = 0;
1511 }
1512
1513 static void rb_inc_iter(struct ring_buffer_iter *iter)
1514 {
1515         struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
1516
1517         /*
1518          * The iterator could be on the reader page (it starts there).
1519          * But the head could have moved, since the reader was
1520          * found. Check for this case and assign the iterator
1521          * to the head page instead of next.
1522          */
1523         if (iter->head_page == cpu_buffer->reader_page)
1524                 iter->head_page = rb_set_head_page(cpu_buffer);
1525         else
1526                 rb_inc_page(cpu_buffer, &iter->head_page);
1527
1528         iter->read_stamp = iter->head_page->page->time_stamp;
1529         iter->head = 0;
1530 }
1531
1532 /**
1533  * ring_buffer_update_event - update event type and data
1534  * @event: the even to update
1535  * @type: the type of event
1536  * @length: the size of the event field in the ring buffer
1537  *
1538  * Update the type and data fields of the event. The length
1539  * is the actual size that is written to the ring buffer,
1540  * and with this, we can determine what to place into the
1541  * data field.
1542  */
1543 static void
1544 rb_update_event(struct ring_buffer_event *event,
1545                          unsigned type, unsigned length)
1546 {
1547         event->type_len = type;
1548
1549         switch (type) {
1550
1551         case RINGBUF_TYPE_PADDING:
1552         case RINGBUF_TYPE_TIME_EXTEND:
1553         case RINGBUF_TYPE_TIME_STAMP:
1554                 break;
1555
1556         case 0:
1557                 length -= RB_EVNT_HDR_SIZE;
1558                 if (length > RB_MAX_SMALL_DATA || RB_FORCE_8BYTE_ALIGNMENT)
1559                         event->array[0] = length;
1560                 else
1561                         event->type_len = DIV_ROUND_UP(length, RB_ALIGNMENT);
1562                 break;
1563         default:
1564                 BUG();
1565         }
1566 }
1567
1568 /*
1569  * rb_handle_head_page - writer hit the head page
1570  *
1571  * Returns: +1 to retry page
1572  *           0 to continue
1573  *          -1 on error
1574  */
1575 static int
1576 rb_handle_head_page(struct ring_buffer_per_cpu *cpu_buffer,
1577                     struct buffer_page *tail_page,
1578                     struct buffer_page *next_page)
1579 {
1580         struct buffer_page *new_head;
1581         int entries;
1582         int type;
1583         int ret;
1584
1585         entries = rb_page_entries(next_page);
1586
1587         /*
1588          * The hard part is here. We need to move the head
1589          * forward, and protect against both readers on
1590          * other CPUs and writers coming in via interrupts.
1591          */
1592         type = rb_head_page_set_update(cpu_buffer, next_page, tail_page,
1593                                        RB_PAGE_HEAD);
1594
1595         /*
1596          * type can be one of four:
1597          *  NORMAL - an interrupt already moved it for us
1598          *  HEAD   - we are the first to get here.
1599          *  UPDATE - we are the interrupt interrupting
1600          *           a current move.
1601          *  MOVED  - a reader on another CPU moved the next
1602          *           pointer to its reader page. Give up
1603          *           and try again.
1604          */
1605
1606         switch (type) {
1607         case RB_PAGE_HEAD:
1608                 /*
1609                  * We changed the head to UPDATE, thus
1610                  * it is our responsibility to update
1611                  * the counters.
1612                  */
1613                 local_add(entries, &cpu_buffer->overrun);
1614
1615                 /*
1616                  * The entries will be zeroed out when we move the
1617                  * tail page.
1618                  */
1619
1620                 /* still more to do */
1621                 break;
1622
1623         case RB_PAGE_UPDATE:
1624                 /*
1625                  * This is an interrupt that interrupt the
1626                  * previous update. Still more to do.
1627                  */
1628                 break;
1629         case RB_PAGE_NORMAL:
1630                 /*
1631                  * An interrupt came in before the update
1632                  * and processed this for us.
1633                  * Nothing left to do.
1634                  */
1635                 return 1;
1636         case RB_PAGE_MOVED:
1637                 /*
1638                  * The reader is on another CPU and just did
1639                  * a swap with our next_page.
1640                  * Try again.
1641                  */
1642                 return 1;
1643         default:
1644                 RB_WARN_ON(cpu_buffer, 1); /* WTF??? */
1645                 return -1;
1646         }
1647
1648         /*
1649          * Now that we are here, the old head pointer is
1650          * set to UPDATE. This will keep the reader from
1651          * swapping the head page with the reader page.
1652          * The reader (on another CPU) will spin till
1653          * we are finished.
1654          *
1655          * We just need to protect against interrupts
1656          * doing the job. We will set the next pointer
1657          * to HEAD. After that, we set the old pointer
1658          * to NORMAL, but only if it was HEAD before.
1659          * otherwise we are an interrupt, and only
1660          * want the outer most commit to reset it.
1661          */
1662         new_head = next_page;
1663         rb_inc_page(cpu_buffer, &new_head);
1664
1665         ret = rb_head_page_set_head(cpu_buffer, new_head, next_page,
1666                                     RB_PAGE_NORMAL);
1667
1668         /*
1669          * Valid returns are:
1670          *  HEAD   - an interrupt came in and already set it.
1671          *  NORMAL - One of two things:
1672          *            1) We really set it.
1673          *            2) A bunch of interrupts came in and moved
1674          *               the page forward again.
1675          */
1676         switch (ret) {
1677         case RB_PAGE_HEAD:
1678         case RB_PAGE_NORMAL:
1679                 /* OK */
1680                 break;
1681         default:
1682                 RB_WARN_ON(cpu_buffer, 1);
1683                 return -1;
1684         }
1685
1686         /*
1687          * It is possible that an interrupt came in,
1688          * set the head up, then more interrupts came in
1689          * and moved it again. When we get back here,
1690          * the page would have been set to NORMAL but we
1691          * just set it back to HEAD.
1692          *
1693          * How do you detect this? Well, if that happened
1694          * the tail page would have moved.
1695          */
1696         if (ret == RB_PAGE_NORMAL) {
1697                 /*
1698                  * If the tail had moved passed next, then we need
1699                  * to reset the pointer.
1700                  */
1701                 if (cpu_buffer->tail_page != tail_page &&
1702                     cpu_buffer->tail_page != next_page)
1703                         rb_head_page_set_normal(cpu_buffer, new_head,
1704                                                 next_page,
1705                                                 RB_PAGE_HEAD);
1706         }
1707
1708         /*
1709          * If this was the outer most commit (the one that
1710          * changed the original pointer from HEAD to UPDATE),
1711          * then it is up to us to reset it to NORMAL.
1712          */
1713         if (type == RB_PAGE_HEAD) {
1714                 ret = rb_head_page_set_normal(cpu_buffer, next_page,
1715                                               tail_page,
1716                                               RB_PAGE_UPDATE);
1717                 if (RB_WARN_ON(cpu_buffer,
1718                                ret != RB_PAGE_UPDATE))
1719                         return -1;
1720         }
1721
1722         return 0;
1723 }
1724
1725 static unsigned rb_calculate_event_length(unsigned length)
1726 {
1727         struct ring_buffer_event event; /* Used only for sizeof array */
1728
1729         /* zero length can cause confusions */
1730         if (!length)
1731                 length = 1;
1732
1733         if (length > RB_MAX_SMALL_DATA || RB_FORCE_8BYTE_ALIGNMENT)
1734                 length += sizeof(event.array[0]);
1735
1736         length += RB_EVNT_HDR_SIZE;
1737         length = ALIGN(length, RB_ARCH_ALIGNMENT);
1738
1739         return length;
1740 }
1741
1742 static inline void
1743 rb_reset_tail(struct ring_buffer_per_cpu *cpu_buffer,
1744               struct buffer_page *tail_page,
1745               unsigned long tail, unsigned long length)
1746 {
1747         struct ring_buffer_event *event;
1748
1749         /*
1750          * Only the event that crossed the page boundary
1751          * must fill the old tail_page with padding.
1752          */
1753         if (tail >= BUF_PAGE_SIZE) {
1754                 local_sub(length, &tail_page->write);
1755                 return;
1756         }
1757
1758         event = __rb_page_index(tail_page, tail);
1759         kmemcheck_annotate_bitfield(event, bitfield);
1760
1761         /*
1762          * If this event is bigger than the minimum size, then
1763          * we need to be careful that we don't subtract the
1764          * write counter enough to allow another writer to slip
1765          * in on this page.
1766          * We put in a discarded commit instead, to make sure
1767          * that this space is not used again.
1768          *
1769          * If we are less than the minimum size, we don't need to
1770          * worry about it.
1771          */
1772         if (tail > (BUF_PAGE_SIZE - RB_EVNT_MIN_SIZE)) {
1773                 /* No room for any events */
1774
1775                 /* Mark the rest of the page with padding */
1776                 rb_event_set_padding(event);
1777
1778                 /* Set the write back to the previous setting */
1779                 local_sub(length, &tail_page->write);
1780                 return;
1781         }
1782
1783         /* Put in a discarded event */
1784         event->array[0] = (BUF_PAGE_SIZE - tail) - RB_EVNT_HDR_SIZE;
1785         event->type_len = RINGBUF_TYPE_PADDING;
1786         /* time delta must be non zero */
1787         event->time_delta = 1;
1788
1789         /* Set write to end of buffer */
1790         length = (tail + length) - BUF_PAGE_SIZE;
1791         local_sub(length, &tail_page->write);
1792 }
1793
1794 static struct ring_buffer_event *
1795 rb_move_tail(struct ring_buffer_per_cpu *cpu_buffer,
1796              unsigned long length, unsigned long tail,
1797              struct buffer_page *tail_page, u64 *ts)
1798 {
1799         struct buffer_page *commit_page = cpu_buffer->commit_page;
1800         struct ring_buffer *buffer = cpu_buffer->buffer;
1801         struct buffer_page *next_page;
1802         int ret;
1803
1804         next_page = tail_page;
1805
1806         rb_inc_page(cpu_buffer, &next_page);
1807
1808         /*
1809          * If for some reason, we had an interrupt storm that made
1810          * it all the way around the buffer, bail, and warn
1811          * about it.
1812          */
1813         if (unlikely(next_page == commit_page)) {
1814                 local_inc(&cpu_buffer->commit_overrun);
1815                 goto out_reset;
1816         }
1817
1818         /*
1819          * This is where the fun begins!
1820          *
1821          * We are fighting against races between a reader that
1822          * could be on another CPU trying to swap its reader
1823          * page with the buffer head.
1824          *
1825          * We are also fighting against interrupts coming in and
1826          * moving the head or tail on us as well.
1827          *
1828          * If the next page is the head page then we have filled
1829          * the buffer, unless the commit page is still on the
1830          * reader page.
1831          */
1832         if (rb_is_head_page(cpu_buffer, next_page, &tail_page->list)) {
1833
1834                 /*
1835                  * If the commit is not on the reader page, then
1836                  * move the header page.
1837                  */
1838                 if (!rb_is_reader_page(cpu_buffer->commit_page)) {
1839                         /*
1840                          * If we are not in overwrite mode,
1841                          * this is easy, just stop here.
1842                          */
1843                         if (!(buffer->flags & RB_FL_OVERWRITE))
1844                                 goto out_reset;
1845
1846                         ret = rb_handle_head_page(cpu_buffer,
1847                                                   tail_page,
1848                                                   next_page);
1849                         if (ret < 0)
1850                                 goto out_reset;
1851                         if (ret)
1852                                 goto out_again;
1853                 } else {
1854                         /*
1855                          * We need to be careful here too. The
1856                          * commit page could still be on the reader
1857                          * page. We could have a small buffer, and
1858                          * have filled up the buffer with events
1859                          * from interrupts and such, and wrapped.
1860                          *
1861                          * Note, if the tail page is also the on the
1862                          * reader_page, we let it move out.
1863                          */
1864                         if (unlikely((cpu_buffer->commit_page !=
1865                                       cpu_buffer->tail_page) &&
1866                                      (cpu_buffer->commit_page ==
1867                                       cpu_buffer->reader_page))) {
1868                                 local_inc(&cpu_buffer->commit_overrun);
1869                                 goto out_reset;
1870                         }
1871                 }
1872         }
1873
1874         ret = rb_tail_page_update(cpu_buffer, tail_page, next_page);
1875         if (ret) {
1876                 /*
1877                  * Nested commits always have zero deltas, so
1878                  * just reread the time stamp
1879                  */
1880                 *ts = rb_time_stamp(buffer);
1881                 next_page->page->time_stamp = *ts;
1882         }
1883
1884  out_again:
1885
1886         rb_reset_tail(cpu_buffer, tail_page, tail, length);
1887
1888         /* fail and let the caller try again */
1889         return ERR_PTR(-EAGAIN);
1890
1891  out_reset:
1892         /* reset write */
1893         rb_reset_tail(cpu_buffer, tail_page, tail, length);
1894
1895         return NULL;
1896 }
1897
1898 static struct ring_buffer_event *
1899 __rb_reserve_next(struct ring_buffer_per_cpu *cpu_buffer,
1900                   unsigned type, unsigned long length, u64 *ts)
1901 {
1902         struct buffer_page *tail_page;
1903         struct ring_buffer_event *event;
1904         unsigned long tail, write;
1905
1906         tail_page = cpu_buffer->tail_page;
1907         write = local_add_return(length, &tail_page->write);
1908
1909         /* set write to only the index of the write */
1910         write &= RB_WRITE_MASK;
1911         tail = write - length;
1912
1913         /* See if we shot pass the end of this buffer page */
1914         if (write > BUF_PAGE_SIZE)
1915                 return rb_move_tail(cpu_buffer, length, tail,
1916                                     tail_page, ts);
1917
1918         /* We reserved something on the buffer */
1919
1920         event = __rb_page_index(tail_page, tail);
1921         kmemcheck_annotate_bitfield(event, bitfield);
1922         rb_update_event(event, type, length);
1923
1924         /* The passed in type is zero for DATA */
1925         if (likely(!type))
1926                 local_inc(&tail_page->entries);
1927
1928         /*
1929          * If this is the first commit on the page, then update
1930          * its timestamp.
1931          */
1932         if (!tail)
1933                 tail_page->page->time_stamp = *ts;
1934
1935         return event;
1936 }
1937
1938 static inline int
1939 rb_try_to_discard(struct ring_buffer_per_cpu *cpu_buffer,
1940                   struct ring_buffer_event *event)
1941 {
1942         unsigned long new_index, old_index;
1943         struct buffer_page *bpage;
1944         unsigned long index;
1945         unsigned long addr;
1946
1947         new_index = rb_event_index(event);
1948         old_index = new_index + rb_event_length(event);
1949         addr = (unsigned long)event;
1950         addr &= PAGE_MASK;
1951
1952         bpage = cpu_buffer->tail_page;
1953
1954         if (bpage->page == (void *)addr && rb_page_write(bpage) == old_index) {
1955                 unsigned long write_mask =
1956                         local_read(&bpage->write) & ~RB_WRITE_MASK;
1957                 /*
1958                  * This is on the tail page. It is possible that
1959                  * a write could come in and move the tail page
1960                  * and write to the next page. That is fine
1961                  * because we just shorten what is on this page.
1962                  */
1963                 old_index += write_mask;
1964                 new_index += write_mask;
1965                 index = local_cmpxchg(&bpage->write, old_index, new_index);
1966                 if (index == old_index)
1967                         return 1;
1968         }
1969
1970         /* could not discard */
1971         return 0;
1972 }
1973
1974 static int
1975 rb_add_time_stamp(struct ring_buffer_per_cpu *cpu_buffer,
1976                   u64 *ts, u64 *delta)
1977 {
1978         struct ring_buffer_event *event;
1979         static int once;
1980         int ret;
1981
1982         if (unlikely(*delta > (1ULL << 59) && !once++)) {
1983                 printk(KERN_WARNING "Delta way too big! %llu"
1984                        " ts=%llu write stamp = %llu\n",
1985                        (unsigned long long)*delta,
1986                        (unsigned long long)*ts,
1987                        (unsigned long long)cpu_buffer->write_stamp);
1988                 WARN_ON(1);
1989         }
1990
1991         /*
1992          * The delta is too big, we to add a
1993          * new timestamp.
1994          */
1995         event = __rb_reserve_next(cpu_buffer,
1996                                   RINGBUF_TYPE_TIME_EXTEND,
1997                                   RB_LEN_TIME_EXTEND,
1998                                   ts);
1999         if (!event)
2000                 return -EBUSY;
2001
2002         if (PTR_ERR(event) == -EAGAIN)
2003                 return -EAGAIN;
2004
2005         /* Only a commited time event can update the write stamp */
2006         if (rb_event_is_commit(cpu_buffer, event)) {
2007                 /*
2008                  * If this is the first on the page, then it was
2009                  * updated with the page itself. Try to discard it
2010                  * and if we can't just make it zero.
2011                  */
2012                 if (rb_event_index(event)) {
2013                         event->time_delta = *delta & TS_MASK;
2014                         event->array[0] = *delta >> TS_SHIFT;
2015                 } else {
2016                         /* try to discard, since we do not need this */
2017                         if (!rb_try_to_discard(cpu_buffer, event)) {
2018                                 /* nope, just zero it */
2019                                 event->time_delta = 0;
2020                                 event->array[0] = 0;
2021                         }
2022                 }
2023                 cpu_buffer->write_stamp = *ts;
2024                 /* let the caller know this was the commit */
2025                 ret = 1;
2026         } else {
2027                 /* Try to discard the event */
2028                 if (!rb_try_to_discard(cpu_buffer, event)) {
2029                         /* Darn, this is just wasted space */
2030                         event->time_delta = 0;
2031                         event->array[0] = 0;
2032                 }
2033                 ret = 0;
2034         }
2035
2036         *delta = 0;
2037
2038         return ret;
2039 }
2040
2041 static void rb_start_commit(struct ring_buffer_per_cpu *cpu_buffer)
2042 {
2043         local_inc(&cpu_buffer->committing);
2044         local_inc(&cpu_buffer->commits);
2045 }
2046
2047 static void rb_end_commit(struct ring_buffer_per_cpu *cpu_buffer)
2048 {
2049         unsigned long commits;
2050
2051         if (RB_WARN_ON(cpu_buffer,
2052                        !local_read(&cpu_buffer->committing)))
2053                 return;
2054
2055  again:
2056         commits = local_read(&cpu_buffer->commits);
2057         /* synchronize with interrupts */
2058         barrier();
2059         if (local_read(&cpu_buffer->committing) == 1)
2060                 rb_set_commit_to_write(cpu_buffer);
2061
2062         local_dec(&cpu_buffer->committing);
2063
2064         /* synchronize with interrupts */
2065         barrier();
2066
2067         /*
2068          * Need to account for interrupts coming in between the
2069          * updating of the commit page and the clearing of the
2070          * committing counter.
2071          */
2072         if (unlikely(local_read(&cpu_buffer->commits) != commits) &&
2073             !local_read(&cpu_buffer->committing)) {
2074                 local_inc(&cpu_buffer->committing);
2075                 goto again;
2076         }
2077 }
2078
2079 static struct ring_buffer_event *
2080 rb_reserve_next_event(struct ring_buffer *buffer,
2081                       struct ring_buffer_per_cpu *cpu_buffer,
2082                       unsigned long length)
2083 {
2084         struct ring_buffer_event *event;
2085         u64 ts, delta = 0;
2086         int commit = 0;
2087         int nr_loops = 0;
2088
2089         rb_start_commit(cpu_buffer);
2090
2091 #ifdef CONFIG_RING_BUFFER_ALLOW_SWAP
2092         /*
2093          * Due to the ability to swap a cpu buffer from a buffer
2094          * it is possible it was swapped before we committed.
2095          * (committing stops a swap). We check for it here and
2096          * if it happened, we have to fail the write.
2097          */
2098         barrier();
2099         if (unlikely(ACCESS_ONCE(cpu_buffer->buffer) != buffer)) {
2100                 local_dec(&cpu_buffer->committing);
2101                 local_dec(&cpu_buffer->commits);
2102                 return NULL;
2103         }
2104 #endif
2105
2106         length = rb_calculate_event_length(length);
2107  again:
2108         /*
2109          * We allow for interrupts to reenter here and do a trace.
2110          * If one does, it will cause this original code to loop
2111          * back here. Even with heavy interrupts happening, this
2112          * should only happen a few times in a row. If this happens
2113          * 1000 times in a row, there must be either an interrupt
2114          * storm or we have something buggy.
2115          * Bail!
2116          */
2117         if (RB_WARN_ON(cpu_buffer, ++nr_loops > 1000))
2118                 goto out_fail;
2119
2120         ts = rb_time_stamp(cpu_buffer->buffer);
2121
2122         /*
2123          * Only the first commit can update the timestamp.
2124          * Yes there is a race here. If an interrupt comes in
2125          * just after the conditional and it traces too, then it
2126          * will also check the deltas. More than one timestamp may
2127          * also be made. But only the entry that did the actual
2128          * commit will be something other than zero.
2129          */
2130         if (likely(cpu_buffer->tail_page == cpu_buffer->commit_page &&
2131                    rb_page_write(cpu_buffer->tail_page) ==
2132                    rb_commit_index(cpu_buffer))) {
2133                 u64 diff;
2134
2135                 diff = ts - cpu_buffer->write_stamp;
2136
2137                 /* make sure this diff is calculated here */
2138                 barrier();
2139
2140                 /* Did the write stamp get updated already? */
2141                 if (unlikely(ts < cpu_buffer->write_stamp))
2142                         goto get_event;
2143
2144                 delta = diff;
2145                 if (unlikely(test_time_stamp(delta))) {
2146
2147                         commit = rb_add_time_stamp(cpu_buffer, &ts, &delta);
2148                         if (commit == -EBUSY)
2149                                 goto out_fail;
2150
2151                         if (commit == -EAGAIN)
2152                                 goto again;
2153
2154                         RB_WARN_ON(cpu_buffer, commit < 0);
2155                 }
2156         }
2157
2158  get_event:
2159         event = __rb_reserve_next(cpu_buffer, 0, length, &ts);
2160         if (unlikely(PTR_ERR(event) == -EAGAIN))
2161                 goto again;
2162
2163         if (!event)
2164                 goto out_fail;
2165
2166         if (!rb_event_is_commit(cpu_buffer, event))
2167                 delta = 0;
2168
2169         event->time_delta = delta;
2170
2171         return event;
2172
2173  out_fail:
2174         rb_end_commit(cpu_buffer);
2175         return NULL;
2176 }
2177
2178 #ifdef CONFIG_TRACING
2179
2180 #define TRACE_RECURSIVE_DEPTH 16
2181
2182 static int trace_recursive_lock(void)
2183 {
2184         current->trace_recursion++;
2185
2186         if (likely(current->trace_recursion < TRACE_RECURSIVE_DEPTH))
2187                 return 0;
2188
2189         /* Disable all tracing before we do anything else */
2190         tracing_off_permanent();
2191
2192         printk_once(KERN_WARNING "Tracing recursion: depth[%ld]:"
2193                     "HC[%lu]:SC[%lu]:NMI[%lu]\n",
2194                     current->trace_recursion,
2195                     hardirq_count() >> HARDIRQ_SHIFT,
2196                     softirq_count() >> SOFTIRQ_SHIFT,
2197                     in_nmi());
2198
2199         WARN_ON_ONCE(1);
2200         return -1;
2201 }
2202
2203 static void trace_recursive_unlock(void)
2204 {
2205         WARN_ON_ONCE(!current->trace_recursion);
2206
2207         current->trace_recursion--;
2208 }
2209
2210 #else
2211
2212 #define trace_recursive_lock()          (0)
2213 #define trace_recursive_unlock()        do { } while (0)
2214
2215 #endif
2216
2217 static DEFINE_PER_CPU(int, rb_need_resched);
2218
2219 /**
2220  * ring_buffer_lock_reserve - reserve a part of the buffer
2221  * @buffer: the ring buffer to reserve from
2222  * @length: the length of the data to reserve (excluding event header)
2223  *
2224  * Returns a reseverd event on the ring buffer to copy directly to.
2225  * The user of this interface will need to get the body to write into
2226  * and can use the ring_buffer_event_data() interface.
2227  *
2228  * The length is the length of the data needed, not the event length
2229  * which also includes the event header.
2230  *
2231  * Must be paired with ring_buffer_unlock_commit, unless NULL is returned.
2232  * If NULL is returned, then nothing has been allocated or locked.
2233  */
2234 struct ring_buffer_event *
2235 ring_buffer_lock_reserve(struct ring_buffer *buffer, unsigned long length)
2236 {
2237         struct ring_buffer_per_cpu *cpu_buffer;
2238         struct ring_buffer_event *event;
2239         int cpu, resched;
2240
2241         if (ring_buffer_flags != RB_BUFFERS_ON)
2242                 return NULL;
2243
2244         /* If we are tracing schedule, we don't want to recurse */
2245         resched = ftrace_preempt_disable();
2246
2247         if (atomic_read(&buffer->record_disabled))
2248                 goto out_nocheck;
2249
2250         if (trace_recursive_lock())
2251                 goto out_nocheck;
2252
2253         cpu = raw_smp_processor_id();
2254
2255         if (!cpumask_test_cpu(cpu, buffer->cpumask))
2256                 goto out;
2257
2258         cpu_buffer = buffer->buffers[cpu];
2259
2260         if (atomic_read(&cpu_buffer->record_disabled))
2261                 goto out;
2262
2263         if (length > BUF_MAX_DATA_SIZE)
2264                 goto out;
2265
2266         event = rb_reserve_next_event(buffer, cpu_buffer, length);
2267         if (!event)
2268                 goto out;
2269
2270         /*
2271          * Need to store resched state on this cpu.
2272          * Only the first needs to.
2273          */
2274
2275         if (preempt_count() == 1)
2276                 per_cpu(rb_need_resched, cpu) = resched;
2277
2278         return event;
2279
2280  out:
2281         trace_recursive_unlock();
2282
2283  out_nocheck:
2284         ftrace_preempt_enable(resched);
2285         return NULL;
2286 }
2287 EXPORT_SYMBOL_GPL(ring_buffer_lock_reserve);
2288
2289 static void
2290 rb_update_write_stamp(struct ring_buffer_per_cpu *cpu_buffer,
2291                       struct ring_buffer_event *event)
2292 {
2293         /*
2294          * The event first in the commit queue updates the
2295          * time stamp.
2296          */
2297         if (rb_event_is_commit(cpu_buffer, event))
2298                 cpu_buffer->write_stamp += event->time_delta;
2299 }
2300
2301 static void rb_commit(struct ring_buffer_per_cpu *cpu_buffer,
2302                       struct ring_buffer_event *event)
2303 {
2304         local_inc(&cpu_buffer->entries);
2305         rb_update_write_stamp(cpu_buffer, event);
2306         rb_end_commit(cpu_buffer);
2307 }
2308
2309 /**
2310  * ring_buffer_unlock_commit - commit a reserved
2311  * @buffer: The buffer to commit to
2312  * @event: The event pointer to commit.
2313  *
2314  * This commits the data to the ring buffer, and releases any locks held.
2315  *
2316  * Must be paired with ring_buffer_lock_reserve.
2317  */
2318 int ring_buffer_unlock_commit(struct ring_buffer *buffer,
2319                               struct ring_buffer_event *event)
2320 {
2321         struct ring_buffer_per_cpu *cpu_buffer;
2322         int cpu = raw_smp_processor_id();
2323
2324         cpu_buffer = buffer->buffers[cpu];
2325
2326         rb_commit(cpu_buffer, event);
2327
2328         trace_recursive_unlock();
2329
2330         /*
2331          * Only the last preempt count needs to restore preemption.
2332          */
2333         if (preempt_count() == 1)
2334                 ftrace_preempt_enable(per_cpu(rb_need_resched, cpu));
2335         else
2336                 preempt_enable_no_resched_notrace();
2337
2338         return 0;
2339 }
2340 EXPORT_SYMBOL_GPL(ring_buffer_unlock_commit);
2341
2342 static inline void rb_event_discard(struct ring_buffer_event *event)
2343 {
2344         /* array[0] holds the actual length for the discarded event */
2345         event->array[0] = rb_event_data_length(event) - RB_EVNT_HDR_SIZE;
2346         event->type_len = RINGBUF_TYPE_PADDING;
2347         /* time delta must be non zero */
2348         if (!event->time_delta)
2349                 event->time_delta = 1;
2350 }
2351
2352 /*
2353  * Decrement the entries to the page that an event is on.
2354  * The event does not even need to exist, only the pointer
2355  * to the page it is on. This may only be called before the commit
2356  * takes place.
2357  */
2358 static inline void
2359 rb_decrement_entry(struct ring_buffer_per_cpu *cpu_buffer,
2360                    struct ring_buffer_event *event)
2361 {
2362         unsigned long addr = (unsigned long)event;
2363         struct buffer_page *bpage = cpu_buffer->commit_page;
2364         struct buffer_page *start;
2365
2366         addr &= PAGE_MASK;
2367
2368         /* Do the likely case first */
2369         if (likely(bpage->page == (void *)addr)) {
2370                 local_dec(&bpage->entries);
2371                 return;
2372         }
2373
2374         /*
2375          * Because the commit page may be on the reader page we
2376          * start with the next page and check the end loop there.
2377          */
2378         rb_inc_page(cpu_buffer, &bpage);
2379         start = bpage;
2380         do {
2381                 if (bpage->page == (void *)addr) {
2382                         local_dec(&bpage->entries);
2383                         return;
2384                 }
2385                 rb_inc_page(cpu_buffer, &bpage);
2386         } while (bpage != start);
2387
2388         /* commit not part of this buffer?? */
2389         RB_WARN_ON(cpu_buffer, 1);
2390 }
2391
2392 /**
2393  * ring_buffer_commit_discard - discard an event that has not been committed
2394  * @buffer: the ring buffer
2395  * @event: non committed event to discard
2396  *
2397  * Sometimes an event that is in the ring buffer needs to be ignored.
2398  * This function lets the user discard an event in the ring buffer
2399  * and then that event will not be read later.
2400  *
2401  * This function only works if it is called before the the item has been
2402  * committed. It will try to free the event from the ring buffer
2403  * if another event has not been added behind it.
2404  *
2405  * If another event has been added behind it, it will set the event
2406  * up as discarded, and perform the commit.
2407  *
2408  * If this function is called, do not call ring_buffer_unlock_commit on
2409  * the event.
2410  */
2411 void ring_buffer_discard_commit(struct ring_buffer *buffer,
2412                                 struct ring_buffer_event *event)
2413 {
2414         struct ring_buffer_per_cpu *cpu_buffer;
2415         int cpu;
2416
2417         /* The event is discarded regardless */
2418         rb_event_discard(event);
2419
2420         cpu = smp_processor_id();
2421         cpu_buffer = buffer->buffers[cpu];
2422
2423         /*
2424          * This must only be called if the event has not been
2425          * committed yet. Thus we can assume that preemption
2426          * is still disabled.
2427          */
2428         RB_WARN_ON(buffer, !local_read(&cpu_buffer->committing));
2429
2430         rb_decrement_entry(cpu_buffer, event);
2431         if (rb_try_to_discard(cpu_buffer, event))
2432                 goto out;
2433
2434         /*
2435          * The commit is still visible by the reader, so we
2436          * must still update the timestamp.
2437          */
2438         rb_update_write_stamp(cpu_buffer, event);
2439  out:
2440         rb_end_commit(cpu_buffer);
2441
2442         trace_recursive_unlock();
2443
2444         /*
2445          * Only the last preempt count needs to restore preemption.
2446          */
2447         if (preempt_count() == 1)
2448                 ftrace_preempt_enable(per_cpu(rb_need_resched, cpu));
2449         else
2450                 preempt_enable_no_resched_notrace();
2451
2452 }
2453 EXPORT_SYMBOL_GPL(ring_buffer_discard_commit);
2454
2455 /**
2456  * ring_buffer_write - write data to the buffer without reserving
2457  * @buffer: The ring buffer to write to.
2458  * @length: The length of the data being written (excluding the event header)
2459  * @data: The data to write to the buffer.
2460  *
2461  * This is like ring_buffer_lock_reserve and ring_buffer_unlock_commit as
2462  * one function. If you already have the data to write to the buffer, it
2463  * may be easier to simply call this function.
2464  *
2465  * Note, like ring_buffer_lock_reserve, the length is the length of the data
2466  * and not the length of the event which would hold the header.
2467  */
2468 int ring_buffer_write(struct ring_buffer *buffer,
2469                         unsigned long length,
2470                         void *data)
2471 {
2472         struct ring_buffer_per_cpu *cpu_buffer;
2473         struct ring_buffer_event *event;
2474         void *body;
2475         int ret = -EBUSY;
2476         int cpu, resched;
2477
2478         if (ring_buffer_flags != RB_BUFFERS_ON)
2479                 return -EBUSY;
2480
2481         resched = ftrace_preempt_disable();
2482
2483         if (atomic_read(&buffer->record_disabled))
2484                 goto out;
2485
2486         cpu = raw_smp_processor_id();
2487
2488         if (!cpumask_test_cpu(cpu, buffer->cpumask))
2489                 goto out;
2490
2491         cpu_buffer = buffer->buffers[cpu];
2492
2493         if (atomic_read(&cpu_buffer->record_disabled))
2494                 goto out;
2495
2496         if (length > BUF_MAX_DATA_SIZE)
2497                 goto out;
2498
2499         event = rb_reserve_next_event(buffer, cpu_buffer, length);
2500         if (!event)
2501                 goto out;
2502
2503         body = rb_event_data(event);
2504
2505         memcpy(body, data, length);
2506
2507         rb_commit(cpu_buffer, event);
2508
2509         ret = 0;
2510  out:
2511         ftrace_preempt_enable(resched);
2512
2513         return ret;
2514 }
2515 EXPORT_SYMBOL_GPL(ring_buffer_write);
2516
2517 static int rb_per_cpu_empty(struct ring_buffer_per_cpu *cpu_buffer)
2518 {
2519         struct buffer_page *reader = cpu_buffer->reader_page;
2520         struct buffer_page *head = rb_set_head_page(cpu_buffer);
2521         struct buffer_page *commit = cpu_buffer->commit_page;
2522
2523         /* In case of error, head will be NULL */
2524         if (unlikely(!head))
2525                 return 1;
2526
2527         return reader->read == rb_page_commit(reader) &&
2528                 (commit == reader ||
2529                  (commit == head &&
2530                   head->read == rb_page_commit(commit)));
2531 }
2532
2533 /**
2534  * ring_buffer_record_disable - stop all writes into the buffer
2535  * @buffer: The ring buffer to stop writes to.
2536  *
2537  * This prevents all writes to the buffer. Any attempt to write
2538  * to the buffer after this will fail and return NULL.
2539  *
2540  * The caller should call synchronize_sched() after this.
2541  */
2542 void ring_buffer_record_disable(struct ring_buffer *buffer)
2543 {
2544         atomic_inc(&buffer->record_disabled);
2545 }
2546 EXPORT_SYMBOL_GPL(ring_buffer_record_disable);
2547
2548 /**
2549  * ring_buffer_record_enable - enable writes to the buffer
2550  * @buffer: The ring buffer to enable writes
2551  *
2552  * Note, multiple disables will need the same number of enables
2553  * to truly enable the writing (much like preempt_disable).
2554  */
2555 void ring_buffer_record_enable(struct ring_buffer *buffer)
2556 {
2557         atomic_dec(&buffer->record_disabled);
2558 }
2559 EXPORT_SYMBOL_GPL(ring_buffer_record_enable);
2560
2561 /**
2562  * ring_buffer_record_disable_cpu - stop all writes into the cpu_buffer
2563  * @buffer: The ring buffer to stop writes to.
2564  * @cpu: The CPU buffer to stop
2565  *
2566  * This prevents all writes to the buffer. Any attempt to write
2567  * to the buffer after this will fail and return NULL.
2568  *
2569  * The caller should call synchronize_sched() after this.
2570  */
2571 void ring_buffer_record_disable_cpu(struct ring_buffer *buffer, int cpu)
2572 {
2573         struct ring_buffer_per_cpu *cpu_buffer;
2574
2575         if (!cpumask_test_cpu(cpu, buffer->cpumask))
2576                 return;
2577
2578         cpu_buffer = buffer->buffers[cpu];
2579         atomic_inc(&cpu_buffer->record_disabled);
2580 }
2581 EXPORT_SYMBOL_GPL(ring_buffer_record_disable_cpu);
2582
2583 /**
2584  * ring_buffer_record_enable_cpu - enable writes to the buffer
2585  * @buffer: The ring buffer to enable writes
2586  * @cpu: The CPU to enable.
2587  *
2588  * Note, multiple disables will need the same number of enables
2589  * to truly enable the writing (much like preempt_disable).
2590  */
2591 void ring_buffer_record_enable_cpu(struct ring_buffer *buffer, int cpu)
2592 {
2593         struct ring_buffer_per_cpu *cpu_buffer;
2594
2595         if (!cpumask_test_cpu(cpu, buffer->cpumask))
2596                 return;
2597
2598         cpu_buffer = buffer->buffers[cpu];
2599         atomic_dec(&cpu_buffer->record_disabled);
2600 }
2601 EXPORT_SYMBOL_GPL(ring_buffer_record_enable_cpu);
2602
2603 /**
2604  * ring_buffer_entries_cpu - get the number of entries in a cpu buffer
2605  * @buffer: The ring buffer
2606  * @cpu: The per CPU buffer to get the entries from.
2607  */
2608 unsigned long ring_buffer_entries_cpu(struct ring_buffer *buffer, int cpu)
2609 {
2610         struct ring_buffer_per_cpu *cpu_buffer;
2611         unsigned long ret;
2612
2613         if (!cpumask_test_cpu(cpu, buffer->cpumask))
2614                 return 0;
2615
2616         cpu_buffer = buffer->buffers[cpu];
2617         ret = (local_read(&cpu_buffer->entries) - local_read(&cpu_buffer->overrun))
2618                 - cpu_buffer->read;
2619
2620         return ret;
2621 }
2622 EXPORT_SYMBOL_GPL(ring_buffer_entries_cpu);
2623
2624 /**
2625  * ring_buffer_overrun_cpu - get the number of overruns in a cpu_buffer
2626  * @buffer: The ring buffer
2627  * @cpu: The per CPU buffer to get the number of overruns from
2628  */
2629 unsigned long ring_buffer_overrun_cpu(struct ring_buffer *buffer, int cpu)
2630 {
2631         struct ring_buffer_per_cpu *cpu_buffer;
2632         unsigned long ret;
2633
2634         if (!cpumask_test_cpu(cpu, buffer->cpumask))
2635                 return 0;
2636
2637         cpu_buffer = buffer->buffers[cpu];
2638         ret = local_read(&cpu_buffer->overrun);
2639
2640         return ret;
2641 }
2642 EXPORT_SYMBOL_GPL(ring_buffer_overrun_cpu);
2643
2644 /**
2645  * ring_buffer_commit_overrun_cpu - get the number of overruns caused by commits
2646  * @buffer: The ring buffer
2647  * @cpu: The per CPU buffer to get the number of overruns from
2648  */
2649 unsigned long
2650 ring_buffer_commit_overrun_cpu(struct ring_buffer *buffer, int cpu)
2651 {
2652         struct ring_buffer_per_cpu *cpu_buffer;
2653         unsigned long ret;
2654
2655         if (!cpumask_test_cpu(cpu, buffer->cpumask))
2656                 return 0;
2657
2658         cpu_buffer = buffer->buffers[cpu];
2659         ret = local_read(&cpu_buffer->commit_overrun);
2660
2661         return ret;
2662 }
2663 EXPORT_SYMBOL_GPL(ring_buffer_commit_overrun_cpu);
2664
2665 /**
2666  * ring_buffer_entries - get the number of entries in a buffer
2667  * @buffer: The ring buffer
2668  *
2669  * Returns the total number of entries in the ring buffer
2670  * (all CPU entries)
2671  */
2672 unsigned long ring_buffer_entries(struct ring_buffer *buffer)
2673 {
2674         struct ring_buffer_per_cpu *cpu_buffer;
2675         unsigned long entries = 0;
2676         int cpu;
2677
2678         /* if you care about this being correct, lock the buffer */
2679         for_each_buffer_cpu(buffer, cpu) {
2680                 cpu_buffer = buffer->buffers[cpu];
2681                 entries += (local_read(&cpu_buffer->entries) -
2682                             local_read(&cpu_buffer->overrun)) - cpu_buffer->read;
2683         }
2684
2685         return entries;
2686 }
2687 EXPORT_SYMBOL_GPL(ring_buffer_entries);
2688
2689 /**
2690  * ring_buffer_overruns - get the number of overruns in buffer
2691  * @buffer: The ring buffer
2692  *
2693  * Returns the total number of overruns in the ring buffer
2694  * (all CPU entries)
2695  */
2696 unsigned long ring_buffer_overruns(struct ring_buffer *buffer)
2697 {
2698         struct ring_buffer_per_cpu *cpu_buffer;
2699         unsigned long overruns = 0;
2700         int cpu;
2701
2702         /* if you care about this being correct, lock the buffer */
2703         for_each_buffer_cpu(buffer, cpu) {
2704                 cpu_buffer = buffer->buffers[cpu];
2705                 overruns += local_read(&cpu_buffer->overrun);
2706         }
2707
2708         return overruns;
2709 }
2710 EXPORT_SYMBOL_GPL(ring_buffer_overruns);
2711
2712 static void rb_iter_reset(struct ring_buffer_iter *iter)
2713 {
2714         struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
2715
2716         /* Iterator usage is expected to have record disabled */
2717         if (list_empty(&cpu_buffer->reader_page->list)) {
2718                 iter->head_page = rb_set_head_page(cpu_buffer);
2719                 if (unlikely(!iter->head_page))
2720                         return;
2721                 iter->head = iter->head_page->read;
2722         } else {
2723                 iter->head_page = cpu_buffer->reader_page;
2724                 iter->head = cpu_buffer->reader_page->read;
2725         }
2726         if (iter->head)
2727                 iter->read_stamp = cpu_buffer->read_stamp;
2728         else
2729                 iter->read_stamp = iter->head_page->page->time_stamp;
2730         iter->cache_reader_page = cpu_buffer->reader_page;
2731         iter->cache_read = cpu_buffer->read;
2732 }
2733
2734 /**
2735  * ring_buffer_iter_reset - reset an iterator
2736  * @iter: The iterator to reset
2737  *
2738  * Resets the iterator, so that it will start from the beginning
2739  * again.
2740  */
2741 void ring_buffer_iter_reset(struct ring_buffer_iter *iter)
2742 {
2743         struct ring_buffer_per_cpu *cpu_buffer;
2744         unsigned long flags;
2745
2746         if (!iter)
2747                 return;
2748
2749         cpu_buffer = iter->cpu_buffer;
2750
2751         spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
2752         rb_iter_reset(iter);
2753         spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
2754 }
2755 EXPORT_SYMBOL_GPL(ring_buffer_iter_reset);
2756
2757 /**
2758  * ring_buffer_iter_empty - check if an iterator has no more to read
2759  * @iter: The iterator to check
2760  */
2761 int ring_buffer_iter_empty(struct ring_buffer_iter *iter)
2762 {
2763         struct ring_buffer_per_cpu *cpu_buffer;
2764
2765         cpu_buffer = iter->cpu_buffer;
2766
2767         return iter->head_page == cpu_buffer->commit_page &&
2768                 iter->head == rb_commit_index(cpu_buffer);
2769 }
2770 EXPORT_SYMBOL_GPL(ring_buffer_iter_empty);
2771
2772 static void
2773 rb_update_read_stamp(struct ring_buffer_per_cpu *cpu_buffer,
2774                      struct ring_buffer_event *event)
2775 {
2776         u64 delta;
2777
2778         switch (event->type_len) {
2779         case RINGBUF_TYPE_PADDING:
2780                 return;
2781
2782         case RINGBUF_TYPE_TIME_EXTEND:
2783                 delta = event->array[0];
2784                 delta <<= TS_SHIFT;
2785                 delta += event->time_delta;
2786                 cpu_buffer->read_stamp += delta;
2787                 return;
2788
2789         case RINGBUF_TYPE_TIME_STAMP:
2790                 /* FIXME: not implemented */
2791                 return;
2792
2793         case RINGBUF_TYPE_DATA:
2794                 cpu_buffer->read_stamp += event->time_delta;
2795                 return;
2796
2797         default:
2798                 BUG();
2799         }
2800         return;
2801 }
2802
2803 static void
2804 rb_update_iter_read_stamp(struct ring_buffer_iter *iter,
2805                           struct ring_buffer_event *event)
2806 {
2807         u64 delta;
2808
2809         switch (event->type_len) {
2810         case RINGBUF_TYPE_PADDING:
2811                 return;
2812
2813         case RINGBUF_TYPE_TIME_EXTEND:
2814                 delta = event->array[0];
2815                 delta <<= TS_SHIFT;
2816                 delta += event->time_delta;
2817                 iter->read_stamp += delta;
2818                 return;
2819
2820         case RINGBUF_TYPE_TIME_STAMP:
2821                 /* FIXME: not implemented */
2822                 return;
2823
2824         case RINGBUF_TYPE_DATA:
2825                 iter->read_stamp += event->time_delta;
2826                 return;
2827
2828         default:
2829                 BUG();
2830         }
2831         return;
2832 }
2833
2834 static struct buffer_page *
2835 rb_get_reader_page(struct ring_buffer_per_cpu *cpu_buffer)
2836 {
2837         struct buffer_page *reader = NULL;
2838         unsigned long flags;
2839         int nr_loops = 0;
2840         int ret;
2841
2842         local_irq_save(flags);
2843         arch_spin_lock(&cpu_buffer->lock);
2844
2845  again:
2846         /*
2847          * This should normally only loop twice. But because the
2848          * start of the reader inserts an empty page, it causes
2849          * a case where we will loop three times. There should be no
2850          * reason to loop four times (that I know of).
2851          */
2852         if (RB_WARN_ON(cpu_buffer, ++nr_loops > 3)) {
2853                 reader = NULL;
2854                 goto out;
2855         }
2856
2857         reader = cpu_buffer->reader_page;
2858
2859         /* If there's more to read, return this page */
2860         if (cpu_buffer->reader_page->read < rb_page_size(reader))
2861                 goto out;
2862
2863         /* Never should we have an index greater than the size */
2864         if (RB_WARN_ON(cpu_buffer,
2865                        cpu_buffer->reader_page->read > rb_page_size(reader)))
2866                 goto out;
2867
2868         /* check if we caught up to the tail */
2869         reader = NULL;
2870         if (cpu_buffer->commit_page == cpu_buffer->reader_page)
2871                 goto out;
2872
2873         /*
2874          * Reset the reader page to size zero.
2875          */
2876         local_set(&cpu_buffer->reader_page->write, 0);
2877         local_set(&cpu_buffer->reader_page->entries, 0);
2878         local_set(&cpu_buffer->reader_page->page->commit, 0);
2879
2880  spin:
2881         /*
2882          * Splice the empty reader page into the list around the head.
2883          */
2884         reader = rb_set_head_page(cpu_buffer);
2885         cpu_buffer->reader_page->list.next = rb_list_head(reader->list.next);
2886         cpu_buffer->reader_page->list.prev = reader->list.prev;
2887
2888         /*
2889          * cpu_buffer->pages just needs to point to the buffer, it
2890          *  has no specific buffer page to point to. Lets move it out
2891          *  of our way so we don't accidently swap it.
2892          */
2893         cpu_buffer->pages = reader->list.prev;
2894
2895         /* The reader page will be pointing to the new head */
2896         rb_set_list_to_head(cpu_buffer, &cpu_buffer->reader_page->list);
2897
2898         /*
2899          * Here's the tricky part.
2900          *
2901          * We need to move the pointer past the header page.
2902          * But we can only do that if a writer is not currently
2903          * moving it. The page before the header page has the
2904          * flag bit '1' set if it is pointing to the page we want.
2905          * but if the writer is in the process of moving it
2906          * than it will be '2' or already moved '0'.
2907          */
2908
2909         ret = rb_head_page_replace(reader, cpu_buffer->reader_page);
2910
2911         /*
2912          * If we did not convert it, then we must try again.
2913          */
2914         if (!ret)
2915                 goto spin;
2916
2917         /*
2918          * Yeah! We succeeded in replacing the page.
2919          *
2920          * Now make the new head point back to the reader page.
2921          */
2922         rb_list_head(reader->list.next)->prev = &cpu_buffer->reader_page->list;
2923         rb_inc_page(cpu_buffer, &cpu_buffer->head_page);
2924
2925         /* Finally update the reader page to the new head */
2926         cpu_buffer->reader_page = reader;
2927         rb_reset_reader_page(cpu_buffer);
2928
2929         goto again;
2930
2931  out:
2932         arch_spin_unlock(&cpu_buffer->lock);
2933         local_irq_restore(flags);
2934
2935         return reader;
2936 }
2937
2938 static void rb_advance_reader(struct ring_buffer_per_cpu *cpu_buffer)
2939 {
2940         struct ring_buffer_event *event;
2941         struct buffer_page *reader;
2942         unsigned length;
2943
2944         reader = rb_get_reader_page(cpu_buffer);
2945
2946         /* This function should not be called when buffer is empty */
2947         if (RB_WARN_ON(cpu_buffer, !reader))
2948                 return;
2949
2950         event = rb_reader_event(cpu_buffer);
2951
2952         if (event->type_len <= RINGBUF_TYPE_DATA_TYPE_LEN_MAX)
2953                 cpu_buffer->read++;
2954
2955         rb_update_read_stamp(cpu_buffer, event);
2956
2957         length = rb_event_length(event);
2958         cpu_buffer->reader_page->read += length;
2959 }
2960
2961 static void rb_advance_iter(struct ring_buffer_iter *iter)
2962 {
2963         struct ring_buffer *buffer;
2964         struct ring_buffer_per_cpu *cpu_buffer;
2965         struct ring_buffer_event *event;
2966         unsigned length;
2967
2968         cpu_buffer = iter->cpu_buffer;
2969         buffer = cpu_buffer->buffer;
2970
2971         /*
2972          * Check if we are at the end of the buffer.
2973          */
2974         if (iter->head >= rb_page_size(iter->head_page)) {
2975                 /* discarded commits can make the page empty */
2976                 if (iter->head_page == cpu_buffer->commit_page)
2977                         return;
2978                 rb_inc_iter(iter);
2979                 return;
2980         }
2981
2982         event = rb_iter_head_event(iter);
2983
2984         length = rb_event_length(event);
2985
2986         /*
2987          * This should not be called to advance the header if we are
2988          * at the tail of the buffer.
2989          */
2990         if (RB_WARN_ON(cpu_buffer,
2991                        (iter->head_page == cpu_buffer->commit_page) &&
2992                        (iter->head + length > rb_commit_index(cpu_buffer))))
2993                 return;
2994
2995         rb_update_iter_read_stamp(iter, event);
2996
2997         iter->head += length;
2998
2999         /* check for end of page padding */
3000         if ((iter->head >= rb_page_size(iter->head_page)) &&
3001             (iter->head_page != cpu_buffer->commit_page))
3002                 rb_advance_iter(iter);
3003 }
3004
3005 static struct ring_buffer_event *
3006 rb_buffer_peek(struct ring_buffer_per_cpu *cpu_buffer, u64 *ts)
3007 {
3008         struct ring_buffer_event *event;
3009         struct buffer_page *reader;
3010         int nr_loops = 0;
3011
3012  again:
3013         /*
3014          * We repeat when a timestamp is encountered. It is possible
3015          * to get multiple timestamps from an interrupt entering just
3016          * as one timestamp is about to be written, or from discarded
3017          * commits. The most that we can have is the number on a single page.
3018          */
3019         if (RB_WARN_ON(cpu_buffer, ++nr_loops > RB_TIMESTAMPS_PER_PAGE))
3020                 return NULL;
3021
3022         reader = rb_get_reader_page(cpu_buffer);
3023         if (!reader)
3024                 return NULL;
3025
3026         event = rb_reader_event(cpu_buffer);
3027
3028         switch (event->type_len) {
3029         case RINGBUF_TYPE_PADDING:
3030                 if (rb_null_event(event))
3031                         RB_WARN_ON(cpu_buffer, 1);
3032                 /*
3033                  * Because the writer could be discarding every
3034                  * event it creates (which would probably be bad)
3035                  * if we were to go back to "again" then we may never
3036                  * catch up, and will trigger the warn on, or lock
3037                  * the box. Return the padding, and we will release
3038                  * the current locks, and try again.
3039                  */
3040                 return event;
3041
3042         case RINGBUF_TYPE_TIME_EXTEND:
3043                 /* Internal data, OK to advance */
3044                 rb_advance_reader(cpu_buffer);
3045                 goto again;
3046
3047         case RINGBUF_TYPE_TIME_STAMP:
3048                 /* FIXME: not implemented */
3049                 rb_advance_reader(cpu_buffer);
3050                 goto again;
3051
3052         case RINGBUF_TYPE_DATA:
3053                 if (ts) {
3054                         *ts = cpu_buffer->read_stamp + event->time_delta;
3055                         ring_buffer_normalize_time_stamp(cpu_buffer->buffer,
3056                                                          cpu_buffer->cpu, ts);
3057                 }
3058                 return event;
3059
3060         default:
3061                 BUG();
3062         }
3063
3064         return NULL;
3065 }
3066 EXPORT_SYMBOL_GPL(ring_buffer_peek);
3067
3068 static struct ring_buffer_event *
3069 rb_iter_peek(struct ring_buffer_iter *iter, u64 *ts)
3070 {
3071         struct ring_buffer *buffer;
3072         struct ring_buffer_per_cpu *cpu_buffer;
3073         struct ring_buffer_event *event;
3074         int nr_loops = 0;
3075
3076         cpu_buffer = iter->cpu_buffer;
3077         buffer = cpu_buffer->buffer;
3078
3079         /*
3080          * Check if someone performed a consuming read to
3081          * the buffer. A consuming read invalidates the iterator
3082          * and we need to reset the iterator in this case.
3083          */
3084         if (unlikely(iter->cache_read != cpu_buffer->read ||
3085                      iter->cache_reader_page != cpu_buffer->reader_page))
3086                 rb_iter_reset(iter);
3087
3088  again:
3089         if (ring_buffer_iter_empty(iter))
3090                 return NULL;
3091
3092         /*
3093          * We repeat when a timestamp is encountered.
3094          * We can get multiple timestamps by nested interrupts or also
3095          * if filtering is on (discarding commits). Since discarding
3096          * commits can be frequent we can get a lot of timestamps.
3097          * But we limit them by not adding timestamps if they begin
3098          * at the start of a page.
3099          */
3100         if (RB_WARN_ON(cpu_buffer, ++nr_loops > RB_TIMESTAMPS_PER_PAGE))
3101                 return NULL;
3102
3103         if (rb_per_cpu_empty(cpu_buffer))
3104                 return NULL;
3105
3106         if (iter->head >= local_read(&iter->head_page->page->commit)) {
3107                 rb_inc_iter(iter);
3108                 goto again;
3109         }
3110
3111         event = rb_iter_head_event(iter);
3112
3113         switch (event->type_len) {
3114         case RINGBUF_TYPE_PADDING:
3115                 if (rb_null_event(event)) {
3116                         rb_inc_iter(iter);
3117                         goto again;
3118                 }
3119                 rb_advance_iter(iter);
3120                 return event;
3121
3122         case RINGBUF_TYPE_TIME_EXTEND:
3123                 /* Internal data, OK to advance */
3124                 rb_advance_iter(iter);
3125                 goto again;
3126
3127         case RINGBUF_TYPE_TIME_STAMP:
3128                 /* FIXME: not implemented */
3129                 rb_advance_iter(iter);
3130                 goto again;
3131
3132         case RINGBUF_TYPE_DATA:
3133                 if (ts) {
3134                         *ts = iter->read_stamp + event->time_delta;
3135                         ring_buffer_normalize_time_stamp(buffer,
3136                                                          cpu_buffer->cpu, ts);
3137                 }
3138                 return event;
3139
3140         default:
3141                 BUG();
3142         }
3143
3144         return NULL;
3145 }
3146 EXPORT_SYMBOL_GPL(ring_buffer_iter_peek);
3147
3148 static inline int rb_ok_to_lock(void)
3149 {
3150         /*
3151          * If an NMI die dumps out the content of the ring buffer
3152          * do not grab locks. We also permanently disable the ring
3153          * buffer too. A one time deal is all you get from reading
3154          * the ring buffer from an NMI.
3155          */
3156         if (likely(!in_nmi()))
3157                 return 1;
3158
3159         tracing_off_permanent();
3160         return 0;
3161 }
3162
3163 /**
3164  * ring_buffer_peek - peek at the next event to be read
3165  * @buffer: The ring buffer to read
3166  * @cpu: The cpu to peak at
3167  * @ts: The timestamp counter of this event.
3168  *
3169  * This will return the event that will be read next, but does
3170  * not consume the data.
3171  */
3172 struct ring_buffer_event *
3173 ring_buffer_peek(struct ring_buffer *buffer, int cpu, u64 *ts)
3174 {
3175         struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
3176         struct ring_buffer_event *event;
3177         unsigned long flags;
3178         int dolock;
3179
3180         if (!cpumask_test_cpu(cpu, buffer->cpumask))
3181                 return NULL;
3182
3183         dolock = rb_ok_to_lock();
3184  again:
3185         local_irq_save(flags);
3186         if (dolock)
3187                 spin_lock(&cpu_buffer->reader_lock);
3188         event = rb_buffer_peek(cpu_buffer, ts);
3189         if (event && event->type_len == RINGBUF_TYPE_PADDING)
3190                 rb_advance_reader(cpu_buffer);
3191         if (dolock)
3192                 spin_unlock(&cpu_buffer->reader_lock);
3193         local_irq_restore(flags);
3194
3195         if (event && event->type_len == RINGBUF_TYPE_PADDING)
3196                 goto again;
3197
3198         return event;
3199 }
3200
3201 /**
3202  * ring_buffer_iter_peek - peek at the next event to be read
3203  * @iter: The ring buffer iterator
3204  * @ts: The timestamp counter of this event.
3205  *
3206  * This will return the event that will be read next, but does
3207  * not increment the iterator.
3208  */
3209 struct ring_buffer_event *
3210 ring_buffer_iter_peek(struct ring_buffer_iter *iter, u64 *ts)
3211 {
3212         struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
3213         struct ring_buffer_event *event;
3214         unsigned long flags;
3215
3216  again:
3217         spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
3218         event = rb_iter_peek(iter, ts);
3219         spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
3220
3221         if (event && event->type_len == RINGBUF_TYPE_PADDING)
3222                 goto again;
3223
3224         return event;
3225 }
3226
3227 /**
3228  * ring_buffer_consume - return an event and consume it
3229  * @buffer: The ring buffer to get the next event from
3230  *
3231  * Returns the next event in the ring buffer, and that event is consumed.
3232  * Meaning, that sequential reads will keep returning a different event,
3233  * and eventually empty the ring buffer if the producer is slower.
3234  */
3235 struct ring_buffer_event *
3236 ring_buffer_consume(struct ring_buffer *buffer, int cpu, u64 *ts)
3237 {
3238         struct ring_buffer_per_cpu *cpu_buffer;
3239         struct ring_buffer_event *event = NULL;
3240         unsigned long flags;
3241         int dolock;
3242
3243         dolock = rb_ok_to_lock();
3244
3245  again:
3246         /* might be called in atomic */
3247         preempt_disable();
3248
3249         if (!cpumask_test_cpu(cpu, buffer->cpumask))
3250                 goto out;
3251
3252         cpu_buffer = buffer->buffers[cpu];
3253         local_irq_save(flags);
3254         if (dolock)
3255                 spin_lock(&cpu_buffer->reader_lock);
3256
3257         event = rb_buffer_peek(cpu_buffer, ts);
3258         if (event)
3259                 rb_advance_reader(cpu_buffer);
3260
3261         if (dolock)
3262                 spin_unlock(&cpu_buffer->reader_lock);
3263         local_irq_restore(flags);
3264
3265  out:
3266         preempt_enable();
3267
3268         if (event && event->type_len == RINGBUF_TYPE_PADDING)
3269                 goto again;
3270
3271         return event;
3272 }
3273 EXPORT_SYMBOL_GPL(ring_buffer_consume);
3274
3275 /**
3276  * ring_buffer_read_start - start a non consuming read of the buffer
3277  * @buffer: The ring buffer to read from
3278  * @cpu: The cpu buffer to iterate over
3279  *
3280  * This starts up an iteration through the buffer. It also disables
3281  * the recording to the buffer until the reading is finished.
3282  * This prevents the reading from being corrupted. This is not
3283  * a consuming read, so a producer is not expected.
3284  *
3285  * Must be paired with ring_buffer_finish.
3286  */
3287 struct ring_buffer_iter *
3288 ring_buffer_read_start(struct ring_buffer *buffer, int cpu)
3289 {
3290         struct ring_buffer_per_cpu *cpu_buffer;
3291         struct ring_buffer_iter *iter;
3292         unsigned long flags;
3293
3294         if (!cpumask_test_cpu(cpu, buffer->cpumask))
3295                 return NULL;
3296
3297         iter = kmalloc(sizeof(*iter), GFP_KERNEL);
3298         if (!iter)
3299                 return NULL;
3300
3301         cpu_buffer = buffer->buffers[cpu];
3302
3303         iter->cpu_buffer = cpu_buffer;
3304
3305         atomic_inc(&cpu_buffer->record_disabled);
3306         synchronize_sched();
3307
3308         spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
3309         arch_spin_lock(&cpu_buffer->lock);
3310         rb_iter_reset(iter);
3311         arch_spin_unlock(&cpu_buffer->lock);
3312         spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
3313
3314         return iter;
3315 }
3316 EXPORT_SYMBOL_GPL(ring_buffer_read_start);
3317
3318 /**
3319  * ring_buffer_finish - finish reading the iterator of the buffer
3320  * @iter: The iterator retrieved by ring_buffer_start
3321  *
3322  * This re-enables the recording to the buffer, and frees the
3323  * iterator.
3324  */
3325 void
3326 ring_buffer_read_finish(struct ring_buffer_iter *iter)
3327 {
3328         struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
3329
3330         atomic_dec(&cpu_buffer->record_disabled);
3331         kfree(iter);
3332 }
3333 EXPORT_SYMBOL_GPL(ring_buffer_read_finish);
3334
3335 /**
3336  * ring_buffer_read - read the next item in the ring buffer by the iterator
3337  * @iter: The ring buffer iterator
3338  * @ts: The time stamp of the event read.
3339  *
3340  * This reads the next event in the ring buffer and increments the iterator.
3341  */
3342 struct ring_buffer_event *
3343 ring_buffer_read(struct ring_buffer_iter *iter, u64 *ts)
3344 {
3345         struct ring_buffer_event *event;
3346         struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
3347         unsigned long flags;
3348
3349         spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
3350  again:
3351         event = rb_iter_peek(iter, ts);
3352         if (!event)
3353                 goto out;
3354
3355         if (event->type_len == RINGBUF_TYPE_PADDING)
3356                 goto again;
3357
3358         rb_advance_iter(iter);
3359  out:
3360         spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
3361
3362         return event;
3363 }
3364 EXPORT_SYMBOL_GPL(ring_buffer_read);
3365
3366 /**
3367  * ring_buffer_size - return the size of the ring buffer (in bytes)
3368  * @buffer: The ring buffer.
3369  */
3370 unsigned long ring_buffer_size(struct ring_buffer *buffer)
3371 {
3372         return BUF_PAGE_SIZE * buffer->pages;
3373 }
3374 EXPORT_SYMBOL_GPL(ring_buffer_size);
3375
3376 static void
3377 rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer)
3378 {
3379         rb_head_page_deactivate(cpu_buffer);
3380
3381         cpu_buffer->head_page
3382                 = list_entry(cpu_buffer->pages, struct buffer_page, list);
3383         local_set(&cpu_buffer->head_page->write, 0);
3384         local_set(&cpu_buffer->head_page->entries, 0);
3385         local_set(&cpu_buffer->head_page->page->commit, 0);
3386
3387         cpu_buffer->head_page->read = 0;
3388
3389         cpu_buffer->tail_page = cpu_buffer->head_page;
3390         cpu_buffer->commit_page = cpu_buffer->head_page;
3391
3392         INIT_LIST_HEAD(&cpu_buffer->reader_page->list);
3393         local_set(&cpu_buffer->reader_page->write, 0);
3394         local_set(&cpu_buffer->reader_page->entries, 0);
3395         local_set(&cpu_buffer->reader_page->page->commit, 0);
3396         cpu_buffer->reader_page->read = 0;
3397
3398         local_set(&cpu_buffer->commit_overrun, 0);
3399         local_set(&cpu_buffer->overrun, 0);
3400         local_set(&cpu_buffer->entries, 0);
3401         local_set(&cpu_buffer->committing, 0);
3402         local_set(&cpu_buffer->commits, 0);
3403         cpu_buffer->read = 0;
3404
3405         cpu_buffer->write_stamp = 0;
3406         cpu_buffer->read_stamp = 0;
3407
3408         rb_head_page_activate(cpu_buffer);
3409 }
3410
3411 /**
3412  * ring_buffer_reset_cpu - reset a ring buffer per CPU buffer
3413  * @buffer: The ring buffer to reset a per cpu buffer of
3414  * @cpu: The CPU buffer to be reset
3415  */
3416 void ring_buffer_reset_cpu(struct ring_buffer *buffer, int cpu)
3417 {
3418         struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
3419         unsigned long flags;
3420
3421         if (!cpumask_test_cpu(cpu, buffer->cpumask))
3422                 return;
3423
3424         atomic_inc(&cpu_buffer->record_disabled);
3425
3426         spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
3427
3428         if (RB_WARN_ON(cpu_buffer, local_read(&cpu_buffer->committing)))
3429                 goto out;
3430
3431         arch_spin_lock(&cpu_buffer->lock);
3432
3433         rb_reset_cpu(cpu_buffer);
3434
3435         arch_spin_unlock(&cpu_buffer->lock);
3436
3437  out:
3438         spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
3439
3440         atomic_dec(&cpu_buffer->record_disabled);
3441 }
3442 EXPORT_SYMBOL_GPL(ring_buffer_reset_cpu);
3443
3444 /**
3445  * ring_buffer_reset - reset a ring buffer
3446  * @buffer: The ring buffer to reset all cpu buffers
3447  */
3448 void ring_buffer_reset(struct ring_buffer *buffer)
3449 {
3450         int cpu;
3451
3452         for_each_buffer_cpu(buffer, cpu)
3453                 ring_buffer_reset_cpu(buffer, cpu);
3454 }
3455 EXPORT_SYMBOL_GPL(ring_buffer_reset);
3456
3457 /**
3458  * rind_buffer_empty - is the ring buffer empty?
3459  * @buffer: The ring buffer to test
3460  */
3461 int ring_buffer_empty(struct ring_buffer *buffer)
3462 {
3463         struct ring_buffer_per_cpu *cpu_buffer;
3464         unsigned long flags;
3465         int dolock;
3466         int cpu;
3467         int ret;
3468
3469         dolock = rb_ok_to_lock();
3470
3471         /* yes this is racy, but if you don't like the race, lock the buffer */
3472         for_each_buffer_cpu(buffer, cpu) {
3473                 cpu_buffer = buffer->buffers[cpu];
3474                 local_irq_save(flags);
3475                 if (dolock)
3476                         spin_lock(&cpu_buffer->reader_lock);
3477                 ret = rb_per_cpu_empty(cpu_buffer);
3478                 if (dolock)
3479                         spin_unlock(&cpu_buffer->reader_lock);
3480                 local_irq_restore(flags);
3481
3482                 if (!ret)
3483                         return 0;
3484         }
3485
3486         return 1;
3487 }
3488 EXPORT_SYMBOL_GPL(ring_buffer_empty);
3489
3490 /**
3491  * ring_buffer_empty_cpu - is a cpu buffer of a ring buffer empty?
3492  * @buffer: The ring buffer
3493  * @cpu: The CPU buffer to test
3494  */
3495 int ring_buffer_empty_cpu(struct ring_buffer *buffer, int cpu)
3496 {
3497         struct ring_buffer_per_cpu *cpu_buffer;
3498         unsigned long flags;
3499         int dolock;
3500         int ret;
3501
3502         if (!cpumask_test_cpu(cpu, buffer->cpumask))
3503                 return 1;
3504
3505         dolock = rb_ok_to_lock();
3506
3507         cpu_buffer = buffer->buffers[cpu];
3508         local_irq_save(flags);
3509         if (dolock)
3510                 spin_lock(&cpu_buffer->reader_lock);
3511         ret = rb_per_cpu_empty(cpu_buffer);
3512         if (dolock)
3513                 spin_unlock(&cpu_buffer->reader_lock);
3514         local_irq_restore(flags);
3515
3516         return ret;
3517 }
3518 EXPORT_SYMBOL_GPL(ring_buffer_empty_cpu);
3519
3520 #ifdef CONFIG_RING_BUFFER_ALLOW_SWAP
3521 /**
3522  * ring_buffer_swap_cpu - swap a CPU buffer between two ring buffers
3523  * @buffer_a: One buffer to swap with
3524  * @buffer_b: The other buffer to swap with
3525  *
3526  * This function is useful for tracers that want to take a "snapshot"
3527  * of a CPU buffer and has another back up buffer lying around.
3528  * it is expected that the tracer handles the cpu buffer not being
3529  * used at the moment.
3530  */
3531 int ring_buffer_swap_cpu(struct ring_buffer *buffer_a,
3532                          struct ring_buffer *buffer_b, int cpu)
3533 {
3534         struct ring_buffer_per_cpu *cpu_buffer_a;
3535         struct ring_buffer_per_cpu *cpu_buffer_b;
3536         int ret = -EINVAL;
3537
3538         if (!cpumask_test_cpu(cpu, buffer_a->cpumask) ||
3539             !cpumask_test_cpu(cpu, buffer_b->cpumask))
3540                 goto out;
3541
3542         /* At least make sure the two buffers are somewhat the same */
3543         if (buffer_a->pages != buffer_b->pages)
3544                 goto out;
3545
3546         ret = -EAGAIN;
3547
3548         if (ring_buffer_flags != RB_BUFFERS_ON)
3549                 goto out;
3550
3551         if (atomic_read(&buffer_a->record_disabled))
3552                 goto out;
3553
3554         if (atomic_read(&buffer_b->record_disabled))
3555                 goto out;
3556
3557         cpu_buffer_a = buffer_a->buffers[cpu];
3558         cpu_buffer_b = buffer_b->buffers[cpu];
3559
3560         if (atomic_read(&cpu_buffer_a->record_disabled))
3561                 goto out;
3562
3563         if (atomic_read(&cpu_buffer_b->record_disabled))
3564                 goto out;
3565
3566         /*
3567          * We can't do a synchronize_sched here because this
3568          * function can be called in atomic context.
3569          * Normally this will be called from the same CPU as cpu.
3570          * If not it's up to the caller to protect this.
3571          */
3572         atomic_inc(&cpu_buffer_a->record_disabled);
3573         atomic_inc(&cpu_buffer_b->record_disabled);
3574
3575         ret = -EBUSY;
3576         if (local_read(&cpu_buffer_a->committing))
3577                 goto out_dec;
3578         if (local_read(&cpu_buffer_b->committing))
3579                 goto out_dec;
3580
3581         buffer_a->buffers[cpu] = cpu_buffer_b;
3582         buffer_b->buffers[cpu] = cpu_buffer_a;
3583
3584         cpu_buffer_b->buffer = buffer_a;
3585         cpu_buffer_a->buffer = buffer_b;
3586
3587         ret = 0;
3588
3589 out_dec:
3590         atomic_dec(&cpu_buffer_a->record_disabled);
3591         atomic_dec(&cpu_buffer_b->record_disabled);
3592 out:
3593         return ret;
3594 }
3595 EXPORT_SYMBOL_GPL(ring_buffer_swap_cpu);
3596 #endif /* CONFIG_RING_BUFFER_ALLOW_SWAP */
3597
3598 /**
3599  * ring_buffer_alloc_read_page - allocate a page to read from buffer
3600  * @buffer: the buffer to allocate for.
3601  *
3602  * This function is used in conjunction with ring_buffer_read_page.
3603  * When reading a full page from the ring buffer, these functions
3604  * can be used to speed up the process. The calling function should
3605  * allocate a few pages first with this function. Then when it
3606  * needs to get pages from the ring buffer, it passes the result
3607  * of this function into ring_buffer_read_page, which will swap
3608  * the page that was allocated, with the read page of the buffer.
3609  *
3610  * Returns:
3611  *  The page allocated, or NULL on error.
3612  */
3613 void *ring_buffer_alloc_read_page(struct ring_buffer *buffer)
3614 {
3615         struct buffer_data_page *bpage;
3616         unsigned long addr;
3617
3618         addr = __get_free_page(GFP_KERNEL);
3619         if (!addr)
3620                 return NULL;
3621
3622         bpage = (void *)addr;
3623
3624         rb_init_page(bpage);
3625
3626         return bpage;
3627 }
3628 EXPORT_SYMBOL_GPL(ring_buffer_alloc_read_page);
3629
3630 /**
3631  * ring_buffer_free_read_page - free an allocated read page
3632  * @buffer: the buffer the page was allocate for
3633  * @data: the page to free
3634  *
3635  * Free a page allocated from ring_buffer_alloc_read_page.
3636  */
3637 void ring_buffer_free_read_page(struct ring_buffer *buffer, void *data)
3638 {
3639         free_page((unsigned long)data);
3640 }
3641 EXPORT_SYMBOL_GPL(ring_buffer_free_read_page);
3642
3643 /**
3644  * ring_buffer_read_page - extract a page from the ring buffer
3645  * @buffer: buffer to extract from
3646  * @data_page: the page to use allocated from ring_buffer_alloc_read_page
3647  * @len: amount to extract
3648  * @cpu: the cpu of the buffer to extract
3649  * @full: should the extraction only happen when the page is full.
3650  *
3651  * This function will pull out a page from the ring buffer and consume it.
3652  * @data_page must be the address of the variable that was returned
3653  * from ring_buffer_alloc_read_page. This is because the page might be used
3654  * to swap with a page in the ring buffer.
3655  *
3656  * for example:
3657  *      rpage = ring_buffer_alloc_read_page(buffer);
3658  *      if (!rpage)
3659  *              return error;
3660  *      ret = ring_buffer_read_page(buffer, &rpage, len, cpu, 0);
3661  *      if (ret >= 0)
3662  *              process_page(rpage, ret);
3663  *
3664  * When @full is set, the function will not return true unless
3665  * the writer is off the reader page.
3666  *
3667  * Note: it is up to the calling functions to handle sleeps and wakeups.
3668  *  The ring buffer can be used anywhere in the kernel and can not
3669  *  blindly call wake_up. The layer that uses the ring buffer must be
3670  *  responsible for that.
3671  *
3672  * Returns:
3673  *  >=0 if data has been transferred, returns the offset of consumed data.
3674  *  <0 if no data has been transferred.
3675  */
3676 int ring_buffer_read_page(struct ring_buffer *buffer,
3677                           void **data_page, size_t len, int cpu, int full)
3678 {
3679         struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
3680         struct ring_buffer_event *event;
3681         struct buffer_data_page *bpage;
3682         struct buffer_page *reader;
3683         unsigned long flags;
3684         unsigned int commit;
3685         unsigned int read;
3686         u64 save_timestamp;
3687         int ret = -1;
3688
3689         if (!cpumask_test_cpu(cpu, buffer->cpumask))
3690                 goto out;
3691
3692         /*
3693          * If len is not big enough to hold the page header, then
3694          * we can not copy anything.
3695          */
3696         if (len <= BUF_PAGE_HDR_SIZE)
3697                 goto out;
3698
3699         len -= BUF_PAGE_HDR_SIZE;
3700
3701         if (!data_page)
3702                 goto out;
3703
3704         bpage = *data_page;
3705         if (!bpage)
3706                 goto out;
3707
3708         spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
3709
3710         reader = rb_get_reader_page(cpu_buffer);
3711         if (!reader)
3712                 goto out_unlock;
3713
3714         event = rb_reader_event(cpu_buffer);
3715
3716         read = reader->read;
3717         commit = rb_page_commit(reader);
3718
3719         /*
3720          * If this page has been partially read or
3721          * if len is not big enough to read the rest of the page or
3722          * a writer is still on the page, then
3723          * we must copy the data from the page to the buffer.
3724          * Otherwise, we can simply swap the page with the one passed in.
3725          */
3726         if (read || (len < (commit - read)) ||
3727             cpu_buffer->reader_page == cpu_buffer->commit_page) {
3728                 struct buffer_data_page *rpage = cpu_buffer->reader_page->page;
3729                 unsigned int rpos = read;
3730                 unsigned int pos = 0;
3731                 unsigned int size;
3732
3733                 if (full)
3734                         goto out_unlock;
3735
3736                 if (len > (commit - read))
3737                         len = (commit - read);
3738
3739                 size = rb_event_length(event);
3740
3741                 if (len < size)
3742                         goto out_unlock;
3743
3744                 /* save the current timestamp, since the user will need it */
3745                 save_timestamp = cpu_buffer->read_stamp;
3746
3747                 /* Need to copy one event at a time */
3748                 do {
3749                         memcpy(bpage->data + pos, rpage->data + rpos, size);
3750
3751                         len -= size;
3752
3753                         rb_advance_reader(cpu_buffer);
3754                         rpos = reader->read;
3755                         pos += size;
3756
3757                         event = rb_reader_event(cpu_buffer);
3758                         size = rb_event_length(event);
3759                 } while (len > size);
3760
3761                 /* update bpage */
3762                 local_set(&bpage->commit, pos);
3763                 bpage->time_stamp = save_timestamp;
3764
3765                 /* we copied everything to the beginning */
3766                 read = 0;
3767         } else {
3768                 /* update the entry counter */
3769                 cpu_buffer->read += rb_page_entries(reader);
3770
3771                 /* swap the pages */
3772                 rb_init_page(bpage);
3773                 bpage = reader->page;
3774                 reader->page = *data_page;
3775                 local_set(&reader->write, 0);
3776                 local_set(&reader->entries, 0);
3777                 reader->read = 0;
3778                 *data_page = bpage;
3779         }
3780         ret = read;
3781
3782  out_unlock:
3783         spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
3784
3785  out:
3786         return ret;
3787 }
3788 EXPORT_SYMBOL_GPL(ring_buffer_read_page);
3789
3790 #ifdef CONFIG_TRACING
3791 static ssize_t
3792 rb_simple_read(struct file *filp, char __user *ubuf,
3793                size_t cnt, loff_t *ppos)
3794 {
3795         unsigned long *p = filp->private_data;
3796         char buf[64];
3797         int r;
3798
3799         if (test_bit(RB_BUFFERS_DISABLED_BIT, p))
3800                 r = sprintf(buf, "permanently disabled\n");
3801         else
3802                 r = sprintf(buf, "%d\n", test_bit(RB_BUFFERS_ON_BIT, p));
3803
3804         return simple_read_from_buffer(ubuf, cnt, ppos, buf, r);
3805 }
3806
3807 static ssize_t
3808 rb_simple_write(struct file *filp, const char __user *ubuf,
3809                 size_t cnt, loff_t *ppos)
3810 {
3811         unsigned long *p = filp->private_data;
3812         char buf[64];
3813         unsigned long val;
3814         int ret;
3815
3816         if (cnt >= sizeof(buf))
3817                 return -EINVAL;
3818
3819         if (copy_from_user(&buf, ubuf, cnt))
3820                 return -EFAULT;
3821
3822         buf[cnt] = 0;
3823
3824         ret = strict_strtoul(buf, 10, &val);
3825         if (ret < 0)
3826                 return ret;
3827
3828         if (val)
3829                 set_bit(RB_BUFFERS_ON_BIT, p);
3830         else
3831                 clear_bit(RB_BUFFERS_ON_BIT, p);
3832
3833         (*ppos)++;
3834
3835         return cnt;
3836 }
3837
3838 static const struct file_operations rb_simple_fops = {
3839         .open           = tracing_open_generic,
3840         .read           = rb_simple_read,
3841         .write          = rb_simple_write,
3842 };
3843
3844
3845 static __init int rb_init_debugfs(void)
3846 {
3847         struct dentry *d_tracer;
3848
3849         d_tracer = tracing_init_dentry();
3850
3851         trace_create_file("tracing_on", 0644, d_tracer,
3852                             &ring_buffer_flags, &rb_simple_fops);
3853
3854         return 0;
3855 }
3856
3857 fs_initcall(rb_init_debugfs);
3858 #endif
3859
3860 #ifdef CONFIG_HOTPLUG_CPU
3861 static int rb_cpu_notify(struct notifier_block *self,
3862                          unsigned long action, void *hcpu)
3863 {
3864         struct ring_buffer *buffer =
3865                 container_of(self, struct ring_buffer, cpu_notify);
3866         long cpu = (long)hcpu;
3867
3868         switch (action) {
3869         case CPU_UP_PREPARE:
3870         case CPU_UP_PREPARE_FROZEN:
3871                 if (cpumask_test_cpu(cpu, buffer->cpumask))
3872                         return NOTIFY_OK;
3873
3874                 buffer->buffers[cpu] =
3875                         rb_allocate_cpu_buffer(buffer, cpu);
3876                 if (!buffer->buffers[cpu]) {
3877                         WARN(1, "failed to allocate ring buffer on CPU %ld\n",
3878                              cpu);
3879                         return NOTIFY_OK;
3880                 }
3881                 smp_wmb();
3882                 cpumask_set_cpu(cpu, buffer->cpumask);
3883                 break;
3884         case CPU_DOWN_PREPARE:
3885         case CPU_DOWN_PREPARE_FROZEN:
3886                 /*
3887                  * Do nothing.
3888                  *  If we were to free the buffer, then the user would
3889                  *  lose any trace that was in the buffer.
3890                  */
3891                 break;
3892         default:
3893                 break;
3894         }
3895         return NOTIFY_OK;
3896 }
3897 #endif