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