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