Merge branch 'irq-core-for-linus' of git://git.kernel.org/pub/scm/linux/kernel/git...
[linux-2.6] / 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/spinlock.h>
8 #include <linux/debugfs.h>
9 #include <linux/uaccess.h>
10 #include <linux/module.h>
11 #include <linux/percpu.h>
12 #include <linux/mutex.h>
13 #include <linux/sched.h>        /* used for sched_clock() (for now) */
14 #include <linux/init.h>
15 #include <linux/hash.h>
16 #include <linux/list.h>
17 #include <linux/fs.h>
18
19 #include "trace.h"
20
21 /*
22  * A fast way to enable or disable all ring buffers is to
23  * call tracing_on or tracing_off. Turning off the ring buffers
24  * prevents all ring buffers from being recorded to.
25  * Turning this switch on, makes it OK to write to the
26  * ring buffer, if the ring buffer is enabled itself.
27  *
28  * There's three layers that must be on in order to write
29  * to the ring buffer.
30  *
31  * 1) This global flag must be set.
32  * 2) The ring buffer must be enabled for recording.
33  * 3) The per cpu buffer must be enabled for recording.
34  *
35  * In case of an anomaly, this global flag has a bit set that
36  * will permantly disable all ring buffers.
37  */
38
39 /*
40  * Global flag to disable all recording to ring buffers
41  *  This has two bits: ON, DISABLED
42  *
43  *  ON   DISABLED
44  * ---- ----------
45  *   0      0        : ring buffers are off
46  *   1      0        : ring buffers are on
47  *   X      1        : ring buffers are permanently disabled
48  */
49
50 enum {
51         RB_BUFFERS_ON_BIT       = 0,
52         RB_BUFFERS_DISABLED_BIT = 1,
53 };
54
55 enum {
56         RB_BUFFERS_ON           = 1 << RB_BUFFERS_ON_BIT,
57         RB_BUFFERS_DISABLED     = 1 << RB_BUFFERS_DISABLED_BIT,
58 };
59
60 static long ring_buffer_flags __read_mostly = RB_BUFFERS_ON;
61
62 /**
63  * tracing_on - enable all tracing buffers
64  *
65  * This function enables all tracing buffers that may have been
66  * disabled with tracing_off.
67  */
68 void tracing_on(void)
69 {
70         set_bit(RB_BUFFERS_ON_BIT, &ring_buffer_flags);
71 }
72
73 /**
74  * tracing_off - turn off all tracing buffers
75  *
76  * This function stops all tracing buffers from recording data.
77  * It does not disable any overhead the tracers themselves may
78  * be causing. This function simply causes all recording to
79  * the ring buffers to fail.
80  */
81 void tracing_off(void)
82 {
83         clear_bit(RB_BUFFERS_ON_BIT, &ring_buffer_flags);
84 }
85
86 /**
87  * tracing_off_permanent - permanently disable ring buffers
88  *
89  * This function, once called, will disable all ring buffers
90  * permanenty.
91  */
92 void tracing_off_permanent(void)
93 {
94         set_bit(RB_BUFFERS_DISABLED_BIT, &ring_buffer_flags);
95 }
96
97 #include "trace.h"
98
99 /* Up this if you want to test the TIME_EXTENTS and normalization */
100 #define DEBUG_SHIFT 0
101
102 /* FIXME!!! */
103 u64 ring_buffer_time_stamp(int cpu)
104 {
105         u64 time;
106
107         preempt_disable_notrace();
108         /* shift to debug/test normalization and TIME_EXTENTS */
109         time = sched_clock() << DEBUG_SHIFT;
110         preempt_enable_no_resched_notrace();
111
112         return time;
113 }
114
115 void ring_buffer_normalize_time_stamp(int cpu, u64 *ts)
116 {
117         /* Just stupid testing the normalize function and deltas */
118         *ts >>= DEBUG_SHIFT;
119 }
120
121 #define RB_EVNT_HDR_SIZE (sizeof(struct ring_buffer_event))
122 #define RB_ALIGNMENT_SHIFT      2
123 #define RB_ALIGNMENT            (1 << RB_ALIGNMENT_SHIFT)
124 #define RB_MAX_SMALL_DATA       28
125
126 enum {
127         RB_LEN_TIME_EXTEND = 8,
128         RB_LEN_TIME_STAMP = 16,
129 };
130
131 /* inline for ring buffer fast paths */
132 static inline unsigned
133 rb_event_length(struct ring_buffer_event *event)
134 {
135         unsigned length;
136
137         switch (event->type) {
138         case RINGBUF_TYPE_PADDING:
139                 /* undefined */
140                 return -1;
141
142         case RINGBUF_TYPE_TIME_EXTEND:
143                 return RB_LEN_TIME_EXTEND;
144
145         case RINGBUF_TYPE_TIME_STAMP:
146                 return RB_LEN_TIME_STAMP;
147
148         case RINGBUF_TYPE_DATA:
149                 if (event->len)
150                         length = event->len << RB_ALIGNMENT_SHIFT;
151                 else
152                         length = event->array[0];
153                 return length + RB_EVNT_HDR_SIZE;
154         default:
155                 BUG();
156         }
157         /* not hit */
158         return 0;
159 }
160
161 /**
162  * ring_buffer_event_length - return the length of the event
163  * @event: the event to get the length of
164  */
165 unsigned ring_buffer_event_length(struct ring_buffer_event *event)
166 {
167         return rb_event_length(event);
168 }
169
170 /* inline for ring buffer fast paths */
171 static inline void *
172 rb_event_data(struct ring_buffer_event *event)
173 {
174         BUG_ON(event->type != RINGBUF_TYPE_DATA);
175         /* If length is in len field, then array[0] has the data */
176         if (event->len)
177                 return (void *)&event->array[0];
178         /* Otherwise length is in array[0] and array[1] has the data */
179         return (void *)&event->array[1];
180 }
181
182 /**
183  * ring_buffer_event_data - return the data of the event
184  * @event: the event to get the data from
185  */
186 void *ring_buffer_event_data(struct ring_buffer_event *event)
187 {
188         return rb_event_data(event);
189 }
190
191 #define for_each_buffer_cpu(buffer, cpu)                \
192         for_each_cpu_mask(cpu, buffer->cpumask)
193
194 #define TS_SHIFT        27
195 #define TS_MASK         ((1ULL << TS_SHIFT) - 1)
196 #define TS_DELTA_TEST   (~TS_MASK)
197
198 struct buffer_data_page {
199         u64              time_stamp;    /* page time stamp */
200         local_t          commit;        /* write commited index */
201         unsigned char    data[];        /* data of buffer page */
202 };
203
204 struct buffer_page {
205         local_t          write;         /* index for next write */
206         unsigned         read;          /* index for next read */
207         struct list_head list;          /* list of free pages */
208         struct buffer_data_page *page;  /* Actual data page */
209 };
210
211 static void rb_init_page(struct buffer_data_page *bpage)
212 {
213         local_set(&bpage->commit, 0);
214 }
215
216 /*
217  * Also stolen from mm/slob.c. Thanks to Mathieu Desnoyers for pointing
218  * this issue out.
219  */
220 static inline void free_buffer_page(struct buffer_page *bpage)
221 {
222         if (bpage->page)
223                 free_page((unsigned long)bpage->page);
224         kfree(bpage);
225 }
226
227 /*
228  * We need to fit the time_stamp delta into 27 bits.
229  */
230 static inline int test_time_stamp(u64 delta)
231 {
232         if (delta & TS_DELTA_TEST)
233                 return 1;
234         return 0;
235 }
236
237 #define BUF_PAGE_SIZE (PAGE_SIZE - sizeof(struct buffer_data_page))
238
239 /*
240  * head_page == tail_page && head == tail then buffer is empty.
241  */
242 struct ring_buffer_per_cpu {
243         int                             cpu;
244         struct ring_buffer              *buffer;
245         spinlock_t                      reader_lock; /* serialize readers */
246         raw_spinlock_t                  lock;
247         struct lock_class_key           lock_key;
248         struct list_head                pages;
249         struct buffer_page              *head_page;     /* read from head */
250         struct buffer_page              *tail_page;     /* write to tail */
251         struct buffer_page              *commit_page;   /* commited pages */
252         struct buffer_page              *reader_page;
253         unsigned long                   overrun;
254         unsigned long                   entries;
255         u64                             write_stamp;
256         u64                             read_stamp;
257         atomic_t                        record_disabled;
258 };
259
260 struct ring_buffer {
261         unsigned                        pages;
262         unsigned                        flags;
263         int                             cpus;
264         cpumask_t                       cpumask;
265         atomic_t                        record_disabled;
266
267         struct mutex                    mutex;
268
269         struct ring_buffer_per_cpu      **buffers;
270 };
271
272 struct ring_buffer_iter {
273         struct ring_buffer_per_cpu      *cpu_buffer;
274         unsigned long                   head;
275         struct buffer_page              *head_page;
276         u64                             read_stamp;
277 };
278
279 /* buffer may be either ring_buffer or ring_buffer_per_cpu */
280 #define RB_WARN_ON(buffer, cond)                                \
281         ({                                                      \
282                 int _____ret = unlikely(cond);                  \
283                 if (_____ret) {                                 \
284                         atomic_inc(&buffer->record_disabled);   \
285                         WARN_ON(1);                             \
286                 }                                               \
287                 _____ret;                                       \
288         })
289
290 /**
291  * check_pages - integrity check of buffer pages
292  * @cpu_buffer: CPU buffer with pages to test
293  *
294  * As a safty measure we check to make sure the data pages have not
295  * been corrupted.
296  */
297 static int rb_check_pages(struct ring_buffer_per_cpu *cpu_buffer)
298 {
299         struct list_head *head = &cpu_buffer->pages;
300         struct buffer_page *bpage, *tmp;
301
302         if (RB_WARN_ON(cpu_buffer, head->next->prev != head))
303                 return -1;
304         if (RB_WARN_ON(cpu_buffer, head->prev->next != head))
305                 return -1;
306
307         list_for_each_entry_safe(bpage, tmp, head, list) {
308                 if (RB_WARN_ON(cpu_buffer,
309                                bpage->list.next->prev != &bpage->list))
310                         return -1;
311                 if (RB_WARN_ON(cpu_buffer,
312                                bpage->list.prev->next != &bpage->list))
313                         return -1;
314         }
315
316         return 0;
317 }
318
319 static int rb_allocate_pages(struct ring_buffer_per_cpu *cpu_buffer,
320                              unsigned nr_pages)
321 {
322         struct list_head *head = &cpu_buffer->pages;
323         struct buffer_page *bpage, *tmp;
324         unsigned long addr;
325         LIST_HEAD(pages);
326         unsigned i;
327
328         for (i = 0; i < nr_pages; i++) {
329                 bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),
330                                     GFP_KERNEL, cpu_to_node(cpu_buffer->cpu));
331                 if (!bpage)
332                         goto free_pages;
333                 list_add(&bpage->list, &pages);
334
335                 addr = __get_free_page(GFP_KERNEL);
336                 if (!addr)
337                         goto free_pages;
338                 bpage->page = (void *)addr;
339                 rb_init_page(bpage->page);
340         }
341
342         list_splice(&pages, head);
343
344         rb_check_pages(cpu_buffer);
345
346         return 0;
347
348  free_pages:
349         list_for_each_entry_safe(bpage, tmp, &pages, list) {
350                 list_del_init(&bpage->list);
351                 free_buffer_page(bpage);
352         }
353         return -ENOMEM;
354 }
355
356 static struct ring_buffer_per_cpu *
357 rb_allocate_cpu_buffer(struct ring_buffer *buffer, int cpu)
358 {
359         struct ring_buffer_per_cpu *cpu_buffer;
360         struct buffer_page *bpage;
361         unsigned long addr;
362         int ret;
363
364         cpu_buffer = kzalloc_node(ALIGN(sizeof(*cpu_buffer), cache_line_size()),
365                                   GFP_KERNEL, cpu_to_node(cpu));
366         if (!cpu_buffer)
367                 return NULL;
368
369         cpu_buffer->cpu = cpu;
370         cpu_buffer->buffer = buffer;
371         spin_lock_init(&cpu_buffer->reader_lock);
372         cpu_buffer->lock = (raw_spinlock_t)__RAW_SPIN_LOCK_UNLOCKED;
373         INIT_LIST_HEAD(&cpu_buffer->pages);
374
375         bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),
376                             GFP_KERNEL, cpu_to_node(cpu));
377         if (!bpage)
378                 goto fail_free_buffer;
379
380         cpu_buffer->reader_page = bpage;
381         addr = __get_free_page(GFP_KERNEL);
382         if (!addr)
383                 goto fail_free_reader;
384         bpage->page = (void *)addr;
385         rb_init_page(bpage->page);
386
387         INIT_LIST_HEAD(&cpu_buffer->reader_page->list);
388
389         ret = rb_allocate_pages(cpu_buffer, buffer->pages);
390         if (ret < 0)
391                 goto fail_free_reader;
392
393         cpu_buffer->head_page
394                 = list_entry(cpu_buffer->pages.next, struct buffer_page, list);
395         cpu_buffer->tail_page = cpu_buffer->commit_page = cpu_buffer->head_page;
396
397         return cpu_buffer;
398
399  fail_free_reader:
400         free_buffer_page(cpu_buffer->reader_page);
401
402  fail_free_buffer:
403         kfree(cpu_buffer);
404         return NULL;
405 }
406
407 static void rb_free_cpu_buffer(struct ring_buffer_per_cpu *cpu_buffer)
408 {
409         struct list_head *head = &cpu_buffer->pages;
410         struct buffer_page *bpage, *tmp;
411
412         list_del_init(&cpu_buffer->reader_page->list);
413         free_buffer_page(cpu_buffer->reader_page);
414
415         list_for_each_entry_safe(bpage, tmp, head, list) {
416                 list_del_init(&bpage->list);
417                 free_buffer_page(bpage);
418         }
419         kfree(cpu_buffer);
420 }
421
422 /*
423  * Causes compile errors if the struct buffer_page gets bigger
424  * than the struct page.
425  */
426 extern int ring_buffer_page_too_big(void);
427
428 /**
429  * ring_buffer_alloc - allocate a new ring_buffer
430  * @size: the size in bytes that is needed.
431  * @flags: attributes to set for the ring buffer.
432  *
433  * Currently the only flag that is available is the RB_FL_OVERWRITE
434  * flag. This flag means that the buffer will overwrite old data
435  * when the buffer wraps. If this flag is not set, the buffer will
436  * drop data when the tail hits the head.
437  */
438 struct ring_buffer *ring_buffer_alloc(unsigned long size, unsigned flags)
439 {
440         struct ring_buffer *buffer;
441         int bsize;
442         int cpu;
443
444         /* Paranoid! Optimizes out when all is well */
445         if (sizeof(struct buffer_page) > sizeof(struct page))
446                 ring_buffer_page_too_big();
447
448
449         /* keep it in its own cache line */
450         buffer = kzalloc(ALIGN(sizeof(*buffer), cache_line_size()),
451                          GFP_KERNEL);
452         if (!buffer)
453                 return NULL;
454
455         buffer->pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
456         buffer->flags = flags;
457
458         /* need at least two pages */
459         if (buffer->pages == 1)
460                 buffer->pages++;
461
462         buffer->cpumask = cpu_possible_map;
463         buffer->cpus = nr_cpu_ids;
464
465         bsize = sizeof(void *) * nr_cpu_ids;
466         buffer->buffers = kzalloc(ALIGN(bsize, cache_line_size()),
467                                   GFP_KERNEL);
468         if (!buffer->buffers)
469                 goto fail_free_buffer;
470
471         for_each_buffer_cpu(buffer, cpu) {
472                 buffer->buffers[cpu] =
473                         rb_allocate_cpu_buffer(buffer, cpu);
474                 if (!buffer->buffers[cpu])
475                         goto fail_free_buffers;
476         }
477
478         mutex_init(&buffer->mutex);
479
480         return buffer;
481
482  fail_free_buffers:
483         for_each_buffer_cpu(buffer, cpu) {
484                 if (buffer->buffers[cpu])
485                         rb_free_cpu_buffer(buffer->buffers[cpu]);
486         }
487         kfree(buffer->buffers);
488
489  fail_free_buffer:
490         kfree(buffer);
491         return NULL;
492 }
493
494 /**
495  * ring_buffer_free - free a ring buffer.
496  * @buffer: the buffer to free.
497  */
498 void
499 ring_buffer_free(struct ring_buffer *buffer)
500 {
501         int cpu;
502
503         for_each_buffer_cpu(buffer, cpu)
504                 rb_free_cpu_buffer(buffer->buffers[cpu]);
505
506         kfree(buffer);
507 }
508
509 static void rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer);
510
511 static void
512 rb_remove_pages(struct ring_buffer_per_cpu *cpu_buffer, unsigned nr_pages)
513 {
514         struct buffer_page *bpage;
515         struct list_head *p;
516         unsigned i;
517
518         atomic_inc(&cpu_buffer->record_disabled);
519         synchronize_sched();
520
521         for (i = 0; i < nr_pages; i++) {
522                 if (RB_WARN_ON(cpu_buffer, list_empty(&cpu_buffer->pages)))
523                         return;
524                 p = cpu_buffer->pages.next;
525                 bpage = list_entry(p, struct buffer_page, list);
526                 list_del_init(&bpage->list);
527                 free_buffer_page(bpage);
528         }
529         if (RB_WARN_ON(cpu_buffer, list_empty(&cpu_buffer->pages)))
530                 return;
531
532         rb_reset_cpu(cpu_buffer);
533
534         rb_check_pages(cpu_buffer);
535
536         atomic_dec(&cpu_buffer->record_disabled);
537
538 }
539
540 static void
541 rb_insert_pages(struct ring_buffer_per_cpu *cpu_buffer,
542                 struct list_head *pages, unsigned nr_pages)
543 {
544         struct buffer_page *bpage;
545         struct list_head *p;
546         unsigned i;
547
548         atomic_inc(&cpu_buffer->record_disabled);
549         synchronize_sched();
550
551         for (i = 0; i < nr_pages; i++) {
552                 if (RB_WARN_ON(cpu_buffer, list_empty(pages)))
553                         return;
554                 p = pages->next;
555                 bpage = list_entry(p, struct buffer_page, list);
556                 list_del_init(&bpage->list);
557                 list_add_tail(&bpage->list, &cpu_buffer->pages);
558         }
559         rb_reset_cpu(cpu_buffer);
560
561         rb_check_pages(cpu_buffer);
562
563         atomic_dec(&cpu_buffer->record_disabled);
564 }
565
566 /**
567  * ring_buffer_resize - resize the ring buffer
568  * @buffer: the buffer to resize.
569  * @size: the new size.
570  *
571  * The tracer is responsible for making sure that the buffer is
572  * not being used while changing the size.
573  * Note: We may be able to change the above requirement by using
574  *  RCU synchronizations.
575  *
576  * Minimum size is 2 * BUF_PAGE_SIZE.
577  *
578  * Returns -1 on failure.
579  */
580 int ring_buffer_resize(struct ring_buffer *buffer, unsigned long size)
581 {
582         struct ring_buffer_per_cpu *cpu_buffer;
583         unsigned nr_pages, rm_pages, new_pages;
584         struct buffer_page *bpage, *tmp;
585         unsigned long buffer_size;
586         unsigned long addr;
587         LIST_HEAD(pages);
588         int i, cpu;
589
590         /*
591          * Always succeed at resizing a non-existent buffer:
592          */
593         if (!buffer)
594                 return size;
595
596         size = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
597         size *= BUF_PAGE_SIZE;
598         buffer_size = buffer->pages * BUF_PAGE_SIZE;
599
600         /* we need a minimum of two pages */
601         if (size < BUF_PAGE_SIZE * 2)
602                 size = BUF_PAGE_SIZE * 2;
603
604         if (size == buffer_size)
605                 return size;
606
607         mutex_lock(&buffer->mutex);
608
609         nr_pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
610
611         if (size < buffer_size) {
612
613                 /* easy case, just free pages */
614                 if (RB_WARN_ON(buffer, nr_pages >= buffer->pages)) {
615                         mutex_unlock(&buffer->mutex);
616                         return -1;
617                 }
618
619                 rm_pages = buffer->pages - nr_pages;
620
621                 for_each_buffer_cpu(buffer, cpu) {
622                         cpu_buffer = buffer->buffers[cpu];
623                         rb_remove_pages(cpu_buffer, rm_pages);
624                 }
625                 goto out;
626         }
627
628         /*
629          * This is a bit more difficult. We only want to add pages
630          * when we can allocate enough for all CPUs. We do this
631          * by allocating all the pages and storing them on a local
632          * link list. If we succeed in our allocation, then we
633          * add these pages to the cpu_buffers. Otherwise we just free
634          * them all and return -ENOMEM;
635          */
636         if (RB_WARN_ON(buffer, nr_pages <= buffer->pages)) {
637                 mutex_unlock(&buffer->mutex);
638                 return -1;
639         }
640
641         new_pages = nr_pages - buffer->pages;
642
643         for_each_buffer_cpu(buffer, cpu) {
644                 for (i = 0; i < new_pages; i++) {
645                         bpage = kzalloc_node(ALIGN(sizeof(*bpage),
646                                                   cache_line_size()),
647                                             GFP_KERNEL, cpu_to_node(cpu));
648                         if (!bpage)
649                                 goto free_pages;
650                         list_add(&bpage->list, &pages);
651                         addr = __get_free_page(GFP_KERNEL);
652                         if (!addr)
653                                 goto free_pages;
654                         bpage->page = (void *)addr;
655                         rb_init_page(bpage->page);
656                 }
657         }
658
659         for_each_buffer_cpu(buffer, cpu) {
660                 cpu_buffer = buffer->buffers[cpu];
661                 rb_insert_pages(cpu_buffer, &pages, new_pages);
662         }
663
664         if (RB_WARN_ON(buffer, !list_empty(&pages))) {
665                 mutex_unlock(&buffer->mutex);
666                 return -1;
667         }
668
669  out:
670         buffer->pages = nr_pages;
671         mutex_unlock(&buffer->mutex);
672
673         return size;
674
675  free_pages:
676         list_for_each_entry_safe(bpage, tmp, &pages, list) {
677                 list_del_init(&bpage->list);
678                 free_buffer_page(bpage);
679         }
680         mutex_unlock(&buffer->mutex);
681         return -ENOMEM;
682 }
683
684 static inline int rb_null_event(struct ring_buffer_event *event)
685 {
686         return event->type == RINGBUF_TYPE_PADDING;
687 }
688
689 static inline void *
690 __rb_data_page_index(struct buffer_data_page *bpage, unsigned index)
691 {
692         return bpage->data + index;
693 }
694
695 static inline void *__rb_page_index(struct buffer_page *bpage, unsigned index)
696 {
697         return bpage->page->data + index;
698 }
699
700 static inline struct ring_buffer_event *
701 rb_reader_event(struct ring_buffer_per_cpu *cpu_buffer)
702 {
703         return __rb_page_index(cpu_buffer->reader_page,
704                                cpu_buffer->reader_page->read);
705 }
706
707 static inline struct ring_buffer_event *
708 rb_head_event(struct ring_buffer_per_cpu *cpu_buffer)
709 {
710         return __rb_page_index(cpu_buffer->head_page,
711                                cpu_buffer->head_page->read);
712 }
713
714 static inline struct ring_buffer_event *
715 rb_iter_head_event(struct ring_buffer_iter *iter)
716 {
717         return __rb_page_index(iter->head_page, iter->head);
718 }
719
720 static inline unsigned rb_page_write(struct buffer_page *bpage)
721 {
722         return local_read(&bpage->write);
723 }
724
725 static inline unsigned rb_page_commit(struct buffer_page *bpage)
726 {
727         return local_read(&bpage->page->commit);
728 }
729
730 /* Size is determined by what has been commited */
731 static inline unsigned rb_page_size(struct buffer_page *bpage)
732 {
733         return rb_page_commit(bpage);
734 }
735
736 static inline unsigned
737 rb_commit_index(struct ring_buffer_per_cpu *cpu_buffer)
738 {
739         return rb_page_commit(cpu_buffer->commit_page);
740 }
741
742 static inline unsigned rb_head_size(struct ring_buffer_per_cpu *cpu_buffer)
743 {
744         return rb_page_commit(cpu_buffer->head_page);
745 }
746
747 /*
748  * When the tail hits the head and the buffer is in overwrite mode,
749  * the head jumps to the next page and all content on the previous
750  * page is discarded. But before doing so, we update the overrun
751  * variable of the buffer.
752  */
753 static void rb_update_overflow(struct ring_buffer_per_cpu *cpu_buffer)
754 {
755         struct ring_buffer_event *event;
756         unsigned long head;
757
758         for (head = 0; head < rb_head_size(cpu_buffer);
759              head += rb_event_length(event)) {
760
761                 event = __rb_page_index(cpu_buffer->head_page, head);
762                 if (RB_WARN_ON(cpu_buffer, rb_null_event(event)))
763                         return;
764                 /* Only count data entries */
765                 if (event->type != RINGBUF_TYPE_DATA)
766                         continue;
767                 cpu_buffer->overrun++;
768                 cpu_buffer->entries--;
769         }
770 }
771
772 static inline void rb_inc_page(struct ring_buffer_per_cpu *cpu_buffer,
773                                struct buffer_page **bpage)
774 {
775         struct list_head *p = (*bpage)->list.next;
776
777         if (p == &cpu_buffer->pages)
778                 p = p->next;
779
780         *bpage = list_entry(p, struct buffer_page, list);
781 }
782
783 static inline unsigned
784 rb_event_index(struct ring_buffer_event *event)
785 {
786         unsigned long addr = (unsigned long)event;
787
788         return (addr & ~PAGE_MASK) - (PAGE_SIZE - BUF_PAGE_SIZE);
789 }
790
791 static inline int
792 rb_is_commit(struct ring_buffer_per_cpu *cpu_buffer,
793              struct ring_buffer_event *event)
794 {
795         unsigned long addr = (unsigned long)event;
796         unsigned long index;
797
798         index = rb_event_index(event);
799         addr &= PAGE_MASK;
800
801         return cpu_buffer->commit_page->page == (void *)addr &&
802                 rb_commit_index(cpu_buffer) == index;
803 }
804
805 static inline void
806 rb_set_commit_event(struct ring_buffer_per_cpu *cpu_buffer,
807                     struct ring_buffer_event *event)
808 {
809         unsigned long addr = (unsigned long)event;
810         unsigned long index;
811
812         index = rb_event_index(event);
813         addr &= PAGE_MASK;
814
815         while (cpu_buffer->commit_page->page != (void *)addr) {
816                 if (RB_WARN_ON(cpu_buffer,
817                           cpu_buffer->commit_page == cpu_buffer->tail_page))
818                         return;
819                 cpu_buffer->commit_page->page->commit =
820                         cpu_buffer->commit_page->write;
821                 rb_inc_page(cpu_buffer, &cpu_buffer->commit_page);
822                 cpu_buffer->write_stamp =
823                         cpu_buffer->commit_page->page->time_stamp;
824         }
825
826         /* Now set the commit to the event's index */
827         local_set(&cpu_buffer->commit_page->page->commit, index);
828 }
829
830 static inline void
831 rb_set_commit_to_write(struct ring_buffer_per_cpu *cpu_buffer)
832 {
833         /*
834          * We only race with interrupts and NMIs on this CPU.
835          * If we own the commit event, then we can commit
836          * all others that interrupted us, since the interruptions
837          * are in stack format (they finish before they come
838          * back to us). This allows us to do a simple loop to
839          * assign the commit to the tail.
840          */
841  again:
842         while (cpu_buffer->commit_page != cpu_buffer->tail_page) {
843                 cpu_buffer->commit_page->page->commit =
844                         cpu_buffer->commit_page->write;
845                 rb_inc_page(cpu_buffer, &cpu_buffer->commit_page);
846                 cpu_buffer->write_stamp =
847                         cpu_buffer->commit_page->page->time_stamp;
848                 /* add barrier to keep gcc from optimizing too much */
849                 barrier();
850         }
851         while (rb_commit_index(cpu_buffer) !=
852                rb_page_write(cpu_buffer->commit_page)) {
853                 cpu_buffer->commit_page->page->commit =
854                         cpu_buffer->commit_page->write;
855                 barrier();
856         }
857
858         /* again, keep gcc from optimizing */
859         barrier();
860
861         /*
862          * If an interrupt came in just after the first while loop
863          * and pushed the tail page forward, we will be left with
864          * a dangling commit that will never go forward.
865          */
866         if (unlikely(cpu_buffer->commit_page != cpu_buffer->tail_page))
867                 goto again;
868 }
869
870 static void rb_reset_reader_page(struct ring_buffer_per_cpu *cpu_buffer)
871 {
872         cpu_buffer->read_stamp = cpu_buffer->reader_page->page->time_stamp;
873         cpu_buffer->reader_page->read = 0;
874 }
875
876 static inline void rb_inc_iter(struct ring_buffer_iter *iter)
877 {
878         struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
879
880         /*
881          * The iterator could be on the reader page (it starts there).
882          * But the head could have moved, since the reader was
883          * found. Check for this case and assign the iterator
884          * to the head page instead of next.
885          */
886         if (iter->head_page == cpu_buffer->reader_page)
887                 iter->head_page = cpu_buffer->head_page;
888         else
889                 rb_inc_page(cpu_buffer, &iter->head_page);
890
891         iter->read_stamp = iter->head_page->page->time_stamp;
892         iter->head = 0;
893 }
894
895 /**
896  * ring_buffer_update_event - update event type and data
897  * @event: the even to update
898  * @type: the type of event
899  * @length: the size of the event field in the ring buffer
900  *
901  * Update the type and data fields of the event. The length
902  * is the actual size that is written to the ring buffer,
903  * and with this, we can determine what to place into the
904  * data field.
905  */
906 static inline void
907 rb_update_event(struct ring_buffer_event *event,
908                          unsigned type, unsigned length)
909 {
910         event->type = type;
911
912         switch (type) {
913
914         case RINGBUF_TYPE_PADDING:
915                 break;
916
917         case RINGBUF_TYPE_TIME_EXTEND:
918                 event->len =
919                         (RB_LEN_TIME_EXTEND + (RB_ALIGNMENT-1))
920                         >> RB_ALIGNMENT_SHIFT;
921                 break;
922
923         case RINGBUF_TYPE_TIME_STAMP:
924                 event->len =
925                         (RB_LEN_TIME_STAMP + (RB_ALIGNMENT-1))
926                         >> RB_ALIGNMENT_SHIFT;
927                 break;
928
929         case RINGBUF_TYPE_DATA:
930                 length -= RB_EVNT_HDR_SIZE;
931                 if (length > RB_MAX_SMALL_DATA) {
932                         event->len = 0;
933                         event->array[0] = length;
934                 } else
935                         event->len =
936                                 (length + (RB_ALIGNMENT-1))
937                                 >> RB_ALIGNMENT_SHIFT;
938                 break;
939         default:
940                 BUG();
941         }
942 }
943
944 static inline unsigned rb_calculate_event_length(unsigned length)
945 {
946         struct ring_buffer_event event; /* Used only for sizeof array */
947
948         /* zero length can cause confusions */
949         if (!length)
950                 length = 1;
951
952         if (length > RB_MAX_SMALL_DATA)
953                 length += sizeof(event.array[0]);
954
955         length += RB_EVNT_HDR_SIZE;
956         length = ALIGN(length, RB_ALIGNMENT);
957
958         return length;
959 }
960
961 static struct ring_buffer_event *
962 __rb_reserve_next(struct ring_buffer_per_cpu *cpu_buffer,
963                   unsigned type, unsigned long length, u64 *ts)
964 {
965         struct buffer_page *tail_page, *head_page, *reader_page, *commit_page;
966         unsigned long tail, write;
967         struct ring_buffer *buffer = cpu_buffer->buffer;
968         struct ring_buffer_event *event;
969         unsigned long flags;
970
971         commit_page = cpu_buffer->commit_page;
972         /* we just need to protect against interrupts */
973         barrier();
974         tail_page = cpu_buffer->tail_page;
975         write = local_add_return(length, &tail_page->write);
976         tail = write - length;
977
978         /* See if we shot pass the end of this buffer page */
979         if (write > BUF_PAGE_SIZE) {
980                 struct buffer_page *next_page = tail_page;
981
982                 local_irq_save(flags);
983                 __raw_spin_lock(&cpu_buffer->lock);
984
985                 rb_inc_page(cpu_buffer, &next_page);
986
987                 head_page = cpu_buffer->head_page;
988                 reader_page = cpu_buffer->reader_page;
989
990                 /* we grabbed the lock before incrementing */
991                 if (RB_WARN_ON(cpu_buffer, next_page == reader_page))
992                         goto out_unlock;
993
994                 /*
995                  * If for some reason, we had an interrupt storm that made
996                  * it all the way around the buffer, bail, and warn
997                  * about it.
998                  */
999                 if (unlikely(next_page == commit_page)) {
1000                         WARN_ON_ONCE(1);
1001                         goto out_unlock;
1002                 }
1003
1004                 if (next_page == head_page) {
1005                         if (!(buffer->flags & RB_FL_OVERWRITE)) {
1006                                 /* reset write */
1007                                 if (tail <= BUF_PAGE_SIZE)
1008                                         local_set(&tail_page->write, tail);
1009                                 goto out_unlock;
1010                         }
1011
1012                         /* tail_page has not moved yet? */
1013                         if (tail_page == cpu_buffer->tail_page) {
1014                                 /* count overflows */
1015                                 rb_update_overflow(cpu_buffer);
1016
1017                                 rb_inc_page(cpu_buffer, &head_page);
1018                                 cpu_buffer->head_page = head_page;
1019                                 cpu_buffer->head_page->read = 0;
1020                         }
1021                 }
1022
1023                 /*
1024                  * If the tail page is still the same as what we think
1025                  * it is, then it is up to us to update the tail
1026                  * pointer.
1027                  */
1028                 if (tail_page == cpu_buffer->tail_page) {
1029                         local_set(&next_page->write, 0);
1030                         local_set(&next_page->page->commit, 0);
1031                         cpu_buffer->tail_page = next_page;
1032
1033                         /* reread the time stamp */
1034                         *ts = ring_buffer_time_stamp(cpu_buffer->cpu);
1035                         cpu_buffer->tail_page->page->time_stamp = *ts;
1036                 }
1037
1038                 /*
1039                  * The actual tail page has moved forward.
1040                  */
1041                 if (tail < BUF_PAGE_SIZE) {
1042                         /* Mark the rest of the page with padding */
1043                         event = __rb_page_index(tail_page, tail);
1044                         event->type = RINGBUF_TYPE_PADDING;
1045                 }
1046
1047                 if (tail <= BUF_PAGE_SIZE)
1048                         /* Set the write back to the previous setting */
1049                         local_set(&tail_page->write, tail);
1050
1051                 /*
1052                  * If this was a commit entry that failed,
1053                  * increment that too
1054                  */
1055                 if (tail_page == cpu_buffer->commit_page &&
1056                     tail == rb_commit_index(cpu_buffer)) {
1057                         rb_set_commit_to_write(cpu_buffer);
1058                 }
1059
1060                 __raw_spin_unlock(&cpu_buffer->lock);
1061                 local_irq_restore(flags);
1062
1063                 /* fail and let the caller try again */
1064                 return ERR_PTR(-EAGAIN);
1065         }
1066
1067         /* We reserved something on the buffer */
1068
1069         if (RB_WARN_ON(cpu_buffer, write > BUF_PAGE_SIZE))
1070                 return NULL;
1071
1072         event = __rb_page_index(tail_page, tail);
1073         rb_update_event(event, type, length);
1074
1075         /*
1076          * If this is a commit and the tail is zero, then update
1077          * this page's time stamp.
1078          */
1079         if (!tail && rb_is_commit(cpu_buffer, event))
1080                 cpu_buffer->commit_page->page->time_stamp = *ts;
1081
1082         return event;
1083
1084  out_unlock:
1085         __raw_spin_unlock(&cpu_buffer->lock);
1086         local_irq_restore(flags);
1087         return NULL;
1088 }
1089
1090 static int
1091 rb_add_time_stamp(struct ring_buffer_per_cpu *cpu_buffer,
1092                   u64 *ts, u64 *delta)
1093 {
1094         struct ring_buffer_event *event;
1095         static int once;
1096         int ret;
1097
1098         if (unlikely(*delta > (1ULL << 59) && !once++)) {
1099                 printk(KERN_WARNING "Delta way too big! %llu"
1100                        " ts=%llu write stamp = %llu\n",
1101                        (unsigned long long)*delta,
1102                        (unsigned long long)*ts,
1103                        (unsigned long long)cpu_buffer->write_stamp);
1104                 WARN_ON(1);
1105         }
1106
1107         /*
1108          * The delta is too big, we to add a
1109          * new timestamp.
1110          */
1111         event = __rb_reserve_next(cpu_buffer,
1112                                   RINGBUF_TYPE_TIME_EXTEND,
1113                                   RB_LEN_TIME_EXTEND,
1114                                   ts);
1115         if (!event)
1116                 return -EBUSY;
1117
1118         if (PTR_ERR(event) == -EAGAIN)
1119                 return -EAGAIN;
1120
1121         /* Only a commited time event can update the write stamp */
1122         if (rb_is_commit(cpu_buffer, event)) {
1123                 /*
1124                  * If this is the first on the page, then we need to
1125                  * update the page itself, and just put in a zero.
1126                  */
1127                 if (rb_event_index(event)) {
1128                         event->time_delta = *delta & TS_MASK;
1129                         event->array[0] = *delta >> TS_SHIFT;
1130                 } else {
1131                         cpu_buffer->commit_page->page->time_stamp = *ts;
1132                         event->time_delta = 0;
1133                         event->array[0] = 0;
1134                 }
1135                 cpu_buffer->write_stamp = *ts;
1136                 /* let the caller know this was the commit */
1137                 ret = 1;
1138         } else {
1139                 /* Darn, this is just wasted space */
1140                 event->time_delta = 0;
1141                 event->array[0] = 0;
1142                 ret = 0;
1143         }
1144
1145         *delta = 0;
1146
1147         return ret;
1148 }
1149
1150 static struct ring_buffer_event *
1151 rb_reserve_next_event(struct ring_buffer_per_cpu *cpu_buffer,
1152                       unsigned type, unsigned long length)
1153 {
1154         struct ring_buffer_event *event;
1155         u64 ts, delta;
1156         int commit = 0;
1157         int nr_loops = 0;
1158
1159  again:
1160         /*
1161          * We allow for interrupts to reenter here and do a trace.
1162          * If one does, it will cause this original code to loop
1163          * back here. Even with heavy interrupts happening, this
1164          * should only happen a few times in a row. If this happens
1165          * 1000 times in a row, there must be either an interrupt
1166          * storm or we have something buggy.
1167          * Bail!
1168          */
1169         if (RB_WARN_ON(cpu_buffer, ++nr_loops > 1000))
1170                 return NULL;
1171
1172         ts = ring_buffer_time_stamp(cpu_buffer->cpu);
1173
1174         /*
1175          * Only the first commit can update the timestamp.
1176          * Yes there is a race here. If an interrupt comes in
1177          * just after the conditional and it traces too, then it
1178          * will also check the deltas. More than one timestamp may
1179          * also be made. But only the entry that did the actual
1180          * commit will be something other than zero.
1181          */
1182         if (cpu_buffer->tail_page == cpu_buffer->commit_page &&
1183             rb_page_write(cpu_buffer->tail_page) ==
1184             rb_commit_index(cpu_buffer)) {
1185
1186                 delta = ts - cpu_buffer->write_stamp;
1187
1188                 /* make sure this delta is calculated here */
1189                 barrier();
1190
1191                 /* Did the write stamp get updated already? */
1192                 if (unlikely(ts < cpu_buffer->write_stamp))
1193                         delta = 0;
1194
1195                 if (test_time_stamp(delta)) {
1196
1197                         commit = rb_add_time_stamp(cpu_buffer, &ts, &delta);
1198
1199                         if (commit == -EBUSY)
1200                                 return NULL;
1201
1202                         if (commit == -EAGAIN)
1203                                 goto again;
1204
1205                         RB_WARN_ON(cpu_buffer, commit < 0);
1206                 }
1207         } else
1208                 /* Non commits have zero deltas */
1209                 delta = 0;
1210
1211         event = __rb_reserve_next(cpu_buffer, type, length, &ts);
1212         if (PTR_ERR(event) == -EAGAIN)
1213                 goto again;
1214
1215         if (!event) {
1216                 if (unlikely(commit))
1217                         /*
1218                          * Ouch! We needed a timestamp and it was commited. But
1219                          * we didn't get our event reserved.
1220                          */
1221                         rb_set_commit_to_write(cpu_buffer);
1222                 return NULL;
1223         }
1224
1225         /*
1226          * If the timestamp was commited, make the commit our entry
1227          * now so that we will update it when needed.
1228          */
1229         if (commit)
1230                 rb_set_commit_event(cpu_buffer, event);
1231         else if (!rb_is_commit(cpu_buffer, event))
1232                 delta = 0;
1233
1234         event->time_delta = delta;
1235
1236         return event;
1237 }
1238
1239 static DEFINE_PER_CPU(int, rb_need_resched);
1240
1241 /**
1242  * ring_buffer_lock_reserve - reserve a part of the buffer
1243  * @buffer: the ring buffer to reserve from
1244  * @length: the length of the data to reserve (excluding event header)
1245  * @flags: a pointer to save the interrupt flags
1246  *
1247  * Returns a reseverd event on the ring buffer to copy directly to.
1248  * The user of this interface will need to get the body to write into
1249  * and can use the ring_buffer_event_data() interface.
1250  *
1251  * The length is the length of the data needed, not the event length
1252  * which also includes the event header.
1253  *
1254  * Must be paired with ring_buffer_unlock_commit, unless NULL is returned.
1255  * If NULL is returned, then nothing has been allocated or locked.
1256  */
1257 struct ring_buffer_event *
1258 ring_buffer_lock_reserve(struct ring_buffer *buffer,
1259                          unsigned long length,
1260                          unsigned long *flags)
1261 {
1262         struct ring_buffer_per_cpu *cpu_buffer;
1263         struct ring_buffer_event *event;
1264         int cpu, resched;
1265
1266         if (ring_buffer_flags != RB_BUFFERS_ON)
1267                 return NULL;
1268
1269         if (atomic_read(&buffer->record_disabled))
1270                 return NULL;
1271
1272         /* If we are tracing schedule, we don't want to recurse */
1273         resched = ftrace_preempt_disable();
1274
1275         cpu = raw_smp_processor_id();
1276
1277         if (!cpu_isset(cpu, buffer->cpumask))
1278                 goto out;
1279
1280         cpu_buffer = buffer->buffers[cpu];
1281
1282         if (atomic_read(&cpu_buffer->record_disabled))
1283                 goto out;
1284
1285         length = rb_calculate_event_length(length);
1286         if (length > BUF_PAGE_SIZE)
1287                 goto out;
1288
1289         event = rb_reserve_next_event(cpu_buffer, RINGBUF_TYPE_DATA, length);
1290         if (!event)
1291                 goto out;
1292
1293         /*
1294          * Need to store resched state on this cpu.
1295          * Only the first needs to.
1296          */
1297
1298         if (preempt_count() == 1)
1299                 per_cpu(rb_need_resched, cpu) = resched;
1300
1301         return event;
1302
1303  out:
1304         ftrace_preempt_enable(resched);
1305         return NULL;
1306 }
1307
1308 static void rb_commit(struct ring_buffer_per_cpu *cpu_buffer,
1309                       struct ring_buffer_event *event)
1310 {
1311         cpu_buffer->entries++;
1312
1313         /* Only process further if we own the commit */
1314         if (!rb_is_commit(cpu_buffer, event))
1315                 return;
1316
1317         cpu_buffer->write_stamp += event->time_delta;
1318
1319         rb_set_commit_to_write(cpu_buffer);
1320 }
1321
1322 /**
1323  * ring_buffer_unlock_commit - commit a reserved
1324  * @buffer: The buffer to commit to
1325  * @event: The event pointer to commit.
1326  * @flags: the interrupt flags received from ring_buffer_lock_reserve.
1327  *
1328  * This commits the data to the ring buffer, and releases any locks held.
1329  *
1330  * Must be paired with ring_buffer_lock_reserve.
1331  */
1332 int ring_buffer_unlock_commit(struct ring_buffer *buffer,
1333                               struct ring_buffer_event *event,
1334                               unsigned long flags)
1335 {
1336         struct ring_buffer_per_cpu *cpu_buffer;
1337         int cpu = raw_smp_processor_id();
1338
1339         cpu_buffer = buffer->buffers[cpu];
1340
1341         rb_commit(cpu_buffer, event);
1342
1343         /*
1344          * Only the last preempt count needs to restore preemption.
1345          */
1346         if (preempt_count() == 1)
1347                 ftrace_preempt_enable(per_cpu(rb_need_resched, cpu));
1348         else
1349                 preempt_enable_no_resched_notrace();
1350
1351         return 0;
1352 }
1353
1354 /**
1355  * ring_buffer_write - write data to the buffer without reserving
1356  * @buffer: The ring buffer to write to.
1357  * @length: The length of the data being written (excluding the event header)
1358  * @data: The data to write to the buffer.
1359  *
1360  * This is like ring_buffer_lock_reserve and ring_buffer_unlock_commit as
1361  * one function. If you already have the data to write to the buffer, it
1362  * may be easier to simply call this function.
1363  *
1364  * Note, like ring_buffer_lock_reserve, the length is the length of the data
1365  * and not the length of the event which would hold the header.
1366  */
1367 int ring_buffer_write(struct ring_buffer *buffer,
1368                         unsigned long length,
1369                         void *data)
1370 {
1371         struct ring_buffer_per_cpu *cpu_buffer;
1372         struct ring_buffer_event *event;
1373         unsigned long event_length;
1374         void *body;
1375         int ret = -EBUSY;
1376         int cpu, resched;
1377
1378         if (ring_buffer_flags != RB_BUFFERS_ON)
1379                 return -EBUSY;
1380
1381         if (atomic_read(&buffer->record_disabled))
1382                 return -EBUSY;
1383
1384         resched = ftrace_preempt_disable();
1385
1386         cpu = raw_smp_processor_id();
1387
1388         if (!cpu_isset(cpu, buffer->cpumask))
1389                 goto out;
1390
1391         cpu_buffer = buffer->buffers[cpu];
1392
1393         if (atomic_read(&cpu_buffer->record_disabled))
1394                 goto out;
1395
1396         event_length = rb_calculate_event_length(length);
1397         event = rb_reserve_next_event(cpu_buffer,
1398                                       RINGBUF_TYPE_DATA, event_length);
1399         if (!event)
1400                 goto out;
1401
1402         body = rb_event_data(event);
1403
1404         memcpy(body, data, length);
1405
1406         rb_commit(cpu_buffer, event);
1407
1408         ret = 0;
1409  out:
1410         ftrace_preempt_enable(resched);
1411
1412         return ret;
1413 }
1414
1415 static inline int rb_per_cpu_empty(struct ring_buffer_per_cpu *cpu_buffer)
1416 {
1417         struct buffer_page *reader = cpu_buffer->reader_page;
1418         struct buffer_page *head = cpu_buffer->head_page;
1419         struct buffer_page *commit = cpu_buffer->commit_page;
1420
1421         return reader->read == rb_page_commit(reader) &&
1422                 (commit == reader ||
1423                  (commit == head &&
1424                   head->read == rb_page_commit(commit)));
1425 }
1426
1427 /**
1428  * ring_buffer_record_disable - stop all writes into the buffer
1429  * @buffer: The ring buffer to stop writes to.
1430  *
1431  * This prevents all writes to the buffer. Any attempt to write
1432  * to the buffer after this will fail and return NULL.
1433  *
1434  * The caller should call synchronize_sched() after this.
1435  */
1436 void ring_buffer_record_disable(struct ring_buffer *buffer)
1437 {
1438         atomic_inc(&buffer->record_disabled);
1439 }
1440
1441 /**
1442  * ring_buffer_record_enable - enable writes to the buffer
1443  * @buffer: The ring buffer to enable writes
1444  *
1445  * Note, multiple disables will need the same number of enables
1446  * to truely enable the writing (much like preempt_disable).
1447  */
1448 void ring_buffer_record_enable(struct ring_buffer *buffer)
1449 {
1450         atomic_dec(&buffer->record_disabled);
1451 }
1452
1453 /**
1454  * ring_buffer_record_disable_cpu - stop all writes into the cpu_buffer
1455  * @buffer: The ring buffer to stop writes to.
1456  * @cpu: The CPU buffer to stop
1457  *
1458  * This prevents all writes to the buffer. Any attempt to write
1459  * to the buffer after this will fail and return NULL.
1460  *
1461  * The caller should call synchronize_sched() after this.
1462  */
1463 void ring_buffer_record_disable_cpu(struct ring_buffer *buffer, int cpu)
1464 {
1465         struct ring_buffer_per_cpu *cpu_buffer;
1466
1467         if (!cpu_isset(cpu, buffer->cpumask))
1468                 return;
1469
1470         cpu_buffer = buffer->buffers[cpu];
1471         atomic_inc(&cpu_buffer->record_disabled);
1472 }
1473
1474 /**
1475  * ring_buffer_record_enable_cpu - enable writes to the buffer
1476  * @buffer: The ring buffer to enable writes
1477  * @cpu: The CPU to enable.
1478  *
1479  * Note, multiple disables will need the same number of enables
1480  * to truely enable the writing (much like preempt_disable).
1481  */
1482 void ring_buffer_record_enable_cpu(struct ring_buffer *buffer, int cpu)
1483 {
1484         struct ring_buffer_per_cpu *cpu_buffer;
1485
1486         if (!cpu_isset(cpu, buffer->cpumask))
1487                 return;
1488
1489         cpu_buffer = buffer->buffers[cpu];
1490         atomic_dec(&cpu_buffer->record_disabled);
1491 }
1492
1493 /**
1494  * ring_buffer_entries_cpu - get the number of entries in a cpu buffer
1495  * @buffer: The ring buffer
1496  * @cpu: The per CPU buffer to get the entries from.
1497  */
1498 unsigned long ring_buffer_entries_cpu(struct ring_buffer *buffer, int cpu)
1499 {
1500         struct ring_buffer_per_cpu *cpu_buffer;
1501
1502         if (!cpu_isset(cpu, buffer->cpumask))
1503                 return 0;
1504
1505         cpu_buffer = buffer->buffers[cpu];
1506         return cpu_buffer->entries;
1507 }
1508
1509 /**
1510  * ring_buffer_overrun_cpu - get the number of overruns in a cpu_buffer
1511  * @buffer: The ring buffer
1512  * @cpu: The per CPU buffer to get the number of overruns from
1513  */
1514 unsigned long ring_buffer_overrun_cpu(struct ring_buffer *buffer, int cpu)
1515 {
1516         struct ring_buffer_per_cpu *cpu_buffer;
1517
1518         if (!cpu_isset(cpu, buffer->cpumask))
1519                 return 0;
1520
1521         cpu_buffer = buffer->buffers[cpu];
1522         return cpu_buffer->overrun;
1523 }
1524
1525 /**
1526  * ring_buffer_entries - get the number of entries in a buffer
1527  * @buffer: The ring buffer
1528  *
1529  * Returns the total number of entries in the ring buffer
1530  * (all CPU entries)
1531  */
1532 unsigned long ring_buffer_entries(struct ring_buffer *buffer)
1533 {
1534         struct ring_buffer_per_cpu *cpu_buffer;
1535         unsigned long entries = 0;
1536         int cpu;
1537
1538         /* if you care about this being correct, lock the buffer */
1539         for_each_buffer_cpu(buffer, cpu) {
1540                 cpu_buffer = buffer->buffers[cpu];
1541                 entries += cpu_buffer->entries;
1542         }
1543
1544         return entries;
1545 }
1546
1547 /**
1548  * ring_buffer_overrun_cpu - get the number of overruns in buffer
1549  * @buffer: The ring buffer
1550  *
1551  * Returns the total number of overruns in the ring buffer
1552  * (all CPU entries)
1553  */
1554 unsigned long ring_buffer_overruns(struct ring_buffer *buffer)
1555 {
1556         struct ring_buffer_per_cpu *cpu_buffer;
1557         unsigned long overruns = 0;
1558         int cpu;
1559
1560         /* if you care about this being correct, lock the buffer */
1561         for_each_buffer_cpu(buffer, cpu) {
1562                 cpu_buffer = buffer->buffers[cpu];
1563                 overruns += cpu_buffer->overrun;
1564         }
1565
1566         return overruns;
1567 }
1568
1569 static void rb_iter_reset(struct ring_buffer_iter *iter)
1570 {
1571         struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
1572
1573         /* Iterator usage is expected to have record disabled */
1574         if (list_empty(&cpu_buffer->reader_page->list)) {
1575                 iter->head_page = cpu_buffer->head_page;
1576                 iter->head = cpu_buffer->head_page->read;
1577         } else {
1578                 iter->head_page = cpu_buffer->reader_page;
1579                 iter->head = cpu_buffer->reader_page->read;
1580         }
1581         if (iter->head)
1582                 iter->read_stamp = cpu_buffer->read_stamp;
1583         else
1584                 iter->read_stamp = iter->head_page->page->time_stamp;
1585 }
1586
1587 /**
1588  * ring_buffer_iter_reset - reset an iterator
1589  * @iter: The iterator to reset
1590  *
1591  * Resets the iterator, so that it will start from the beginning
1592  * again.
1593  */
1594 void ring_buffer_iter_reset(struct ring_buffer_iter *iter)
1595 {
1596         struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
1597         unsigned long flags;
1598
1599         spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
1600         rb_iter_reset(iter);
1601         spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
1602 }
1603
1604 /**
1605  * ring_buffer_iter_empty - check if an iterator has no more to read
1606  * @iter: The iterator to check
1607  */
1608 int ring_buffer_iter_empty(struct ring_buffer_iter *iter)
1609 {
1610         struct ring_buffer_per_cpu *cpu_buffer;
1611
1612         cpu_buffer = iter->cpu_buffer;
1613
1614         return iter->head_page == cpu_buffer->commit_page &&
1615                 iter->head == rb_commit_index(cpu_buffer);
1616 }
1617
1618 static void
1619 rb_update_read_stamp(struct ring_buffer_per_cpu *cpu_buffer,
1620                      struct ring_buffer_event *event)
1621 {
1622         u64 delta;
1623
1624         switch (event->type) {
1625         case RINGBUF_TYPE_PADDING:
1626                 return;
1627
1628         case RINGBUF_TYPE_TIME_EXTEND:
1629                 delta = event->array[0];
1630                 delta <<= TS_SHIFT;
1631                 delta += event->time_delta;
1632                 cpu_buffer->read_stamp += delta;
1633                 return;
1634
1635         case RINGBUF_TYPE_TIME_STAMP:
1636                 /* FIXME: not implemented */
1637                 return;
1638
1639         case RINGBUF_TYPE_DATA:
1640                 cpu_buffer->read_stamp += event->time_delta;
1641                 return;
1642
1643         default:
1644                 BUG();
1645         }
1646         return;
1647 }
1648
1649 static void
1650 rb_update_iter_read_stamp(struct ring_buffer_iter *iter,
1651                           struct ring_buffer_event *event)
1652 {
1653         u64 delta;
1654
1655         switch (event->type) {
1656         case RINGBUF_TYPE_PADDING:
1657                 return;
1658
1659         case RINGBUF_TYPE_TIME_EXTEND:
1660                 delta = event->array[0];
1661                 delta <<= TS_SHIFT;
1662                 delta += event->time_delta;
1663                 iter->read_stamp += delta;
1664                 return;
1665
1666         case RINGBUF_TYPE_TIME_STAMP:
1667                 /* FIXME: not implemented */
1668                 return;
1669
1670         case RINGBUF_TYPE_DATA:
1671                 iter->read_stamp += event->time_delta;
1672                 return;
1673
1674         default:
1675                 BUG();
1676         }
1677         return;
1678 }
1679
1680 static struct buffer_page *
1681 rb_get_reader_page(struct ring_buffer_per_cpu *cpu_buffer)
1682 {
1683         struct buffer_page *reader = NULL;
1684         unsigned long flags;
1685         int nr_loops = 0;
1686
1687         local_irq_save(flags);
1688         __raw_spin_lock(&cpu_buffer->lock);
1689
1690  again:
1691         /*
1692          * This should normally only loop twice. But because the
1693          * start of the reader inserts an empty page, it causes
1694          * a case where we will loop three times. There should be no
1695          * reason to loop four times (that I know of).
1696          */
1697         if (RB_WARN_ON(cpu_buffer, ++nr_loops > 3)) {
1698                 reader = NULL;
1699                 goto out;
1700         }
1701
1702         reader = cpu_buffer->reader_page;
1703
1704         /* If there's more to read, return this page */
1705         if (cpu_buffer->reader_page->read < rb_page_size(reader))
1706                 goto out;
1707
1708         /* Never should we have an index greater than the size */
1709         if (RB_WARN_ON(cpu_buffer,
1710                        cpu_buffer->reader_page->read > rb_page_size(reader)))
1711                 goto out;
1712
1713         /* check if we caught up to the tail */
1714         reader = NULL;
1715         if (cpu_buffer->commit_page == cpu_buffer->reader_page)
1716                 goto out;
1717
1718         /*
1719          * Splice the empty reader page into the list around the head.
1720          * Reset the reader page to size zero.
1721          */
1722
1723         reader = cpu_buffer->head_page;
1724         cpu_buffer->reader_page->list.next = reader->list.next;
1725         cpu_buffer->reader_page->list.prev = reader->list.prev;
1726
1727         local_set(&cpu_buffer->reader_page->write, 0);
1728         local_set(&cpu_buffer->reader_page->page->commit, 0);
1729
1730         /* Make the reader page now replace the head */
1731         reader->list.prev->next = &cpu_buffer->reader_page->list;
1732         reader->list.next->prev = &cpu_buffer->reader_page->list;
1733
1734         /*
1735          * If the tail is on the reader, then we must set the head
1736          * to the inserted page, otherwise we set it one before.
1737          */
1738         cpu_buffer->head_page = cpu_buffer->reader_page;
1739
1740         if (cpu_buffer->commit_page != reader)
1741                 rb_inc_page(cpu_buffer, &cpu_buffer->head_page);
1742
1743         /* Finally update the reader page to the new head */
1744         cpu_buffer->reader_page = reader;
1745         rb_reset_reader_page(cpu_buffer);
1746
1747         goto again;
1748
1749  out:
1750         __raw_spin_unlock(&cpu_buffer->lock);
1751         local_irq_restore(flags);
1752
1753         return reader;
1754 }
1755
1756 static void rb_advance_reader(struct ring_buffer_per_cpu *cpu_buffer)
1757 {
1758         struct ring_buffer_event *event;
1759         struct buffer_page *reader;
1760         unsigned length;
1761
1762         reader = rb_get_reader_page(cpu_buffer);
1763
1764         /* This function should not be called when buffer is empty */
1765         if (RB_WARN_ON(cpu_buffer, !reader))
1766                 return;
1767
1768         event = rb_reader_event(cpu_buffer);
1769
1770         if (event->type == RINGBUF_TYPE_DATA)
1771                 cpu_buffer->entries--;
1772
1773         rb_update_read_stamp(cpu_buffer, event);
1774
1775         length = rb_event_length(event);
1776         cpu_buffer->reader_page->read += length;
1777 }
1778
1779 static void rb_advance_iter(struct ring_buffer_iter *iter)
1780 {
1781         struct ring_buffer *buffer;
1782         struct ring_buffer_per_cpu *cpu_buffer;
1783         struct ring_buffer_event *event;
1784         unsigned length;
1785
1786         cpu_buffer = iter->cpu_buffer;
1787         buffer = cpu_buffer->buffer;
1788
1789         /*
1790          * Check if we are at the end of the buffer.
1791          */
1792         if (iter->head >= rb_page_size(iter->head_page)) {
1793                 if (RB_WARN_ON(buffer,
1794                                iter->head_page == cpu_buffer->commit_page))
1795                         return;
1796                 rb_inc_iter(iter);
1797                 return;
1798         }
1799
1800         event = rb_iter_head_event(iter);
1801
1802         length = rb_event_length(event);
1803
1804         /*
1805          * This should not be called to advance the header if we are
1806          * at the tail of the buffer.
1807          */
1808         if (RB_WARN_ON(cpu_buffer,
1809                        (iter->head_page == cpu_buffer->commit_page) &&
1810                        (iter->head + length > rb_commit_index(cpu_buffer))))
1811                 return;
1812
1813         rb_update_iter_read_stamp(iter, event);
1814
1815         iter->head += length;
1816
1817         /* check for end of page padding */
1818         if ((iter->head >= rb_page_size(iter->head_page)) &&
1819             (iter->head_page != cpu_buffer->commit_page))
1820                 rb_advance_iter(iter);
1821 }
1822
1823 static struct ring_buffer_event *
1824 rb_buffer_peek(struct ring_buffer *buffer, int cpu, u64 *ts)
1825 {
1826         struct ring_buffer_per_cpu *cpu_buffer;
1827         struct ring_buffer_event *event;
1828         struct buffer_page *reader;
1829         int nr_loops = 0;
1830
1831         if (!cpu_isset(cpu, buffer->cpumask))
1832                 return NULL;
1833
1834         cpu_buffer = buffer->buffers[cpu];
1835
1836  again:
1837         /*
1838          * We repeat when a timestamp is encountered. It is possible
1839          * to get multiple timestamps from an interrupt entering just
1840          * as one timestamp is about to be written. The max times
1841          * that this can happen is the number of nested interrupts we
1842          * can have.  Nesting 10 deep of interrupts is clearly
1843          * an anomaly.
1844          */
1845         if (RB_WARN_ON(cpu_buffer, ++nr_loops > 10))
1846                 return NULL;
1847
1848         reader = rb_get_reader_page(cpu_buffer);
1849         if (!reader)
1850                 return NULL;
1851
1852         event = rb_reader_event(cpu_buffer);
1853
1854         switch (event->type) {
1855         case RINGBUF_TYPE_PADDING:
1856                 RB_WARN_ON(cpu_buffer, 1);
1857                 rb_advance_reader(cpu_buffer);
1858                 return NULL;
1859
1860         case RINGBUF_TYPE_TIME_EXTEND:
1861                 /* Internal data, OK to advance */
1862                 rb_advance_reader(cpu_buffer);
1863                 goto again;
1864
1865         case RINGBUF_TYPE_TIME_STAMP:
1866                 /* FIXME: not implemented */
1867                 rb_advance_reader(cpu_buffer);
1868                 goto again;
1869
1870         case RINGBUF_TYPE_DATA:
1871                 if (ts) {
1872                         *ts = cpu_buffer->read_stamp + event->time_delta;
1873                         ring_buffer_normalize_time_stamp(cpu_buffer->cpu, ts);
1874                 }
1875                 return event;
1876
1877         default:
1878                 BUG();
1879         }
1880
1881         return NULL;
1882 }
1883
1884 static struct ring_buffer_event *
1885 rb_iter_peek(struct ring_buffer_iter *iter, u64 *ts)
1886 {
1887         struct ring_buffer *buffer;
1888         struct ring_buffer_per_cpu *cpu_buffer;
1889         struct ring_buffer_event *event;
1890         int nr_loops = 0;
1891
1892         if (ring_buffer_iter_empty(iter))
1893                 return NULL;
1894
1895         cpu_buffer = iter->cpu_buffer;
1896         buffer = cpu_buffer->buffer;
1897
1898  again:
1899         /*
1900          * We repeat when a timestamp is encountered. It is possible
1901          * to get multiple timestamps from an interrupt entering just
1902          * as one timestamp is about to be written. The max times
1903          * that this can happen is the number of nested interrupts we
1904          * can have. Nesting 10 deep of interrupts is clearly
1905          * an anomaly.
1906          */
1907         if (RB_WARN_ON(cpu_buffer, ++nr_loops > 10))
1908                 return NULL;
1909
1910         if (rb_per_cpu_empty(cpu_buffer))
1911                 return NULL;
1912
1913         event = rb_iter_head_event(iter);
1914
1915         switch (event->type) {
1916         case RINGBUF_TYPE_PADDING:
1917                 rb_inc_iter(iter);
1918                 goto again;
1919
1920         case RINGBUF_TYPE_TIME_EXTEND:
1921                 /* Internal data, OK to advance */
1922                 rb_advance_iter(iter);
1923                 goto again;
1924
1925         case RINGBUF_TYPE_TIME_STAMP:
1926                 /* FIXME: not implemented */
1927                 rb_advance_iter(iter);
1928                 goto again;
1929
1930         case RINGBUF_TYPE_DATA:
1931                 if (ts) {
1932                         *ts = iter->read_stamp + event->time_delta;
1933                         ring_buffer_normalize_time_stamp(cpu_buffer->cpu, ts);
1934                 }
1935                 return event;
1936
1937         default:
1938                 BUG();
1939         }
1940
1941         return NULL;
1942 }
1943
1944 /**
1945  * ring_buffer_peek - peek at the next event to be read
1946  * @buffer: The ring buffer to read
1947  * @cpu: The cpu to peak at
1948  * @ts: The timestamp counter of this event.
1949  *
1950  * This will return the event that will be read next, but does
1951  * not consume the data.
1952  */
1953 struct ring_buffer_event *
1954 ring_buffer_peek(struct ring_buffer *buffer, int cpu, u64 *ts)
1955 {
1956         struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
1957         struct ring_buffer_event *event;
1958         unsigned long flags;
1959
1960         spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
1961         event = rb_buffer_peek(buffer, cpu, ts);
1962         spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
1963
1964         return event;
1965 }
1966
1967 /**
1968  * ring_buffer_iter_peek - peek at the next event to be read
1969  * @iter: The ring buffer iterator
1970  * @ts: The timestamp counter of this event.
1971  *
1972  * This will return the event that will be read next, but does
1973  * not increment the iterator.
1974  */
1975 struct ring_buffer_event *
1976 ring_buffer_iter_peek(struct ring_buffer_iter *iter, u64 *ts)
1977 {
1978         struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
1979         struct ring_buffer_event *event;
1980         unsigned long flags;
1981
1982         spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
1983         event = rb_iter_peek(iter, ts);
1984         spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
1985
1986         return event;
1987 }
1988
1989 /**
1990  * ring_buffer_consume - return an event and consume it
1991  * @buffer: The ring buffer to get the next event from
1992  *
1993  * Returns the next event in the ring buffer, and that event is consumed.
1994  * Meaning, that sequential reads will keep returning a different event,
1995  * and eventually empty the ring buffer if the producer is slower.
1996  */
1997 struct ring_buffer_event *
1998 ring_buffer_consume(struct ring_buffer *buffer, int cpu, u64 *ts)
1999 {
2000         struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
2001         struct ring_buffer_event *event;
2002         unsigned long flags;
2003
2004         if (!cpu_isset(cpu, buffer->cpumask))
2005                 return NULL;
2006
2007         spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
2008
2009         event = rb_buffer_peek(buffer, cpu, ts);
2010         if (!event)
2011                 goto out;
2012
2013         rb_advance_reader(cpu_buffer);
2014
2015  out:
2016         spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
2017
2018         return event;
2019 }
2020
2021 /**
2022  * ring_buffer_read_start - start a non consuming read of the buffer
2023  * @buffer: The ring buffer to read from
2024  * @cpu: The cpu buffer to iterate over
2025  *
2026  * This starts up an iteration through the buffer. It also disables
2027  * the recording to the buffer until the reading is finished.
2028  * This prevents the reading from being corrupted. This is not
2029  * a consuming read, so a producer is not expected.
2030  *
2031  * Must be paired with ring_buffer_finish.
2032  */
2033 struct ring_buffer_iter *
2034 ring_buffer_read_start(struct ring_buffer *buffer, int cpu)
2035 {
2036         struct ring_buffer_per_cpu *cpu_buffer;
2037         struct ring_buffer_iter *iter;
2038         unsigned long flags;
2039
2040         if (!cpu_isset(cpu, buffer->cpumask))
2041                 return NULL;
2042
2043         iter = kmalloc(sizeof(*iter), GFP_KERNEL);
2044         if (!iter)
2045                 return NULL;
2046
2047         cpu_buffer = buffer->buffers[cpu];
2048
2049         iter->cpu_buffer = cpu_buffer;
2050
2051         atomic_inc(&cpu_buffer->record_disabled);
2052         synchronize_sched();
2053
2054         spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
2055         __raw_spin_lock(&cpu_buffer->lock);
2056         rb_iter_reset(iter);
2057         __raw_spin_unlock(&cpu_buffer->lock);
2058         spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
2059
2060         return iter;
2061 }
2062
2063 /**
2064  * ring_buffer_finish - finish reading the iterator of the buffer
2065  * @iter: The iterator retrieved by ring_buffer_start
2066  *
2067  * This re-enables the recording to the buffer, and frees the
2068  * iterator.
2069  */
2070 void
2071 ring_buffer_read_finish(struct ring_buffer_iter *iter)
2072 {
2073         struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
2074
2075         atomic_dec(&cpu_buffer->record_disabled);
2076         kfree(iter);
2077 }
2078
2079 /**
2080  * ring_buffer_read - read the next item in the ring buffer by the iterator
2081  * @iter: The ring buffer iterator
2082  * @ts: The time stamp of the event read.
2083  *
2084  * This reads the next event in the ring buffer and increments the iterator.
2085  */
2086 struct ring_buffer_event *
2087 ring_buffer_read(struct ring_buffer_iter *iter, u64 *ts)
2088 {
2089         struct ring_buffer_event *event;
2090         struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
2091         unsigned long flags;
2092
2093         spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
2094         event = rb_iter_peek(iter, ts);
2095         if (!event)
2096                 goto out;
2097
2098         rb_advance_iter(iter);
2099  out:
2100         spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
2101
2102         return event;
2103 }
2104
2105 /**
2106  * ring_buffer_size - return the size of the ring buffer (in bytes)
2107  * @buffer: The ring buffer.
2108  */
2109 unsigned long ring_buffer_size(struct ring_buffer *buffer)
2110 {
2111         return BUF_PAGE_SIZE * buffer->pages;
2112 }
2113
2114 static void
2115 rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer)
2116 {
2117         cpu_buffer->head_page
2118                 = list_entry(cpu_buffer->pages.next, struct buffer_page, list);
2119         local_set(&cpu_buffer->head_page->write, 0);
2120         local_set(&cpu_buffer->head_page->page->commit, 0);
2121
2122         cpu_buffer->head_page->read = 0;
2123
2124         cpu_buffer->tail_page = cpu_buffer->head_page;
2125         cpu_buffer->commit_page = cpu_buffer->head_page;
2126
2127         INIT_LIST_HEAD(&cpu_buffer->reader_page->list);
2128         local_set(&cpu_buffer->reader_page->write, 0);
2129         local_set(&cpu_buffer->reader_page->page->commit, 0);
2130         cpu_buffer->reader_page->read = 0;
2131
2132         cpu_buffer->overrun = 0;
2133         cpu_buffer->entries = 0;
2134 }
2135
2136 /**
2137  * ring_buffer_reset_cpu - reset a ring buffer per CPU buffer
2138  * @buffer: The ring buffer to reset a per cpu buffer of
2139  * @cpu: The CPU buffer to be reset
2140  */
2141 void ring_buffer_reset_cpu(struct ring_buffer *buffer, int cpu)
2142 {
2143         struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
2144         unsigned long flags;
2145
2146         if (!cpu_isset(cpu, buffer->cpumask))
2147                 return;
2148
2149         spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
2150
2151         __raw_spin_lock(&cpu_buffer->lock);
2152
2153         rb_reset_cpu(cpu_buffer);
2154
2155         __raw_spin_unlock(&cpu_buffer->lock);
2156
2157         spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
2158 }
2159
2160 /**
2161  * ring_buffer_reset - reset a ring buffer
2162  * @buffer: The ring buffer to reset all cpu buffers
2163  */
2164 void ring_buffer_reset(struct ring_buffer *buffer)
2165 {
2166         int cpu;
2167
2168         for_each_buffer_cpu(buffer, cpu)
2169                 ring_buffer_reset_cpu(buffer, cpu);
2170 }
2171
2172 /**
2173  * rind_buffer_empty - is the ring buffer empty?
2174  * @buffer: The ring buffer to test
2175  */
2176 int ring_buffer_empty(struct ring_buffer *buffer)
2177 {
2178         struct ring_buffer_per_cpu *cpu_buffer;
2179         int cpu;
2180
2181         /* yes this is racy, but if you don't like the race, lock the buffer */
2182         for_each_buffer_cpu(buffer, cpu) {
2183                 cpu_buffer = buffer->buffers[cpu];
2184                 if (!rb_per_cpu_empty(cpu_buffer))
2185                         return 0;
2186         }
2187         return 1;
2188 }
2189
2190 /**
2191  * ring_buffer_empty_cpu - is a cpu buffer of a ring buffer empty?
2192  * @buffer: The ring buffer
2193  * @cpu: The CPU buffer to test
2194  */
2195 int ring_buffer_empty_cpu(struct ring_buffer *buffer, int cpu)
2196 {
2197         struct ring_buffer_per_cpu *cpu_buffer;
2198
2199         if (!cpu_isset(cpu, buffer->cpumask))
2200                 return 1;
2201
2202         cpu_buffer = buffer->buffers[cpu];
2203         return rb_per_cpu_empty(cpu_buffer);
2204 }
2205
2206 /**
2207  * ring_buffer_swap_cpu - swap a CPU buffer between two ring buffers
2208  * @buffer_a: One buffer to swap with
2209  * @buffer_b: The other buffer to swap with
2210  *
2211  * This function is useful for tracers that want to take a "snapshot"
2212  * of a CPU buffer and has another back up buffer lying around.
2213  * it is expected that the tracer handles the cpu buffer not being
2214  * used at the moment.
2215  */
2216 int ring_buffer_swap_cpu(struct ring_buffer *buffer_a,
2217                          struct ring_buffer *buffer_b, int cpu)
2218 {
2219         struct ring_buffer_per_cpu *cpu_buffer_a;
2220         struct ring_buffer_per_cpu *cpu_buffer_b;
2221
2222         if (!cpu_isset(cpu, buffer_a->cpumask) ||
2223             !cpu_isset(cpu, buffer_b->cpumask))
2224                 return -EINVAL;
2225
2226         /* At least make sure the two buffers are somewhat the same */
2227         if (buffer_a->pages != buffer_b->pages)
2228                 return -EINVAL;
2229
2230         cpu_buffer_a = buffer_a->buffers[cpu];
2231         cpu_buffer_b = buffer_b->buffers[cpu];
2232
2233         /*
2234          * We can't do a synchronize_sched here because this
2235          * function can be called in atomic context.
2236          * Normally this will be called from the same CPU as cpu.
2237          * If not it's up to the caller to protect this.
2238          */
2239         atomic_inc(&cpu_buffer_a->record_disabled);
2240         atomic_inc(&cpu_buffer_b->record_disabled);
2241
2242         buffer_a->buffers[cpu] = cpu_buffer_b;
2243         buffer_b->buffers[cpu] = cpu_buffer_a;
2244
2245         cpu_buffer_b->buffer = buffer_a;
2246         cpu_buffer_a->buffer = buffer_b;
2247
2248         atomic_dec(&cpu_buffer_a->record_disabled);
2249         atomic_dec(&cpu_buffer_b->record_disabled);
2250
2251         return 0;
2252 }
2253
2254 static void rb_remove_entries(struct ring_buffer_per_cpu *cpu_buffer,
2255                               struct buffer_data_page *bpage)
2256 {
2257         struct ring_buffer_event *event;
2258         unsigned long head;
2259
2260         __raw_spin_lock(&cpu_buffer->lock);
2261         for (head = 0; head < local_read(&bpage->commit);
2262              head += rb_event_length(event)) {
2263
2264                 event = __rb_data_page_index(bpage, head);
2265                 if (RB_WARN_ON(cpu_buffer, rb_null_event(event)))
2266                         return;
2267                 /* Only count data entries */
2268                 if (event->type != RINGBUF_TYPE_DATA)
2269                         continue;
2270                 cpu_buffer->entries--;
2271         }
2272         __raw_spin_unlock(&cpu_buffer->lock);
2273 }
2274
2275 /**
2276  * ring_buffer_alloc_read_page - allocate a page to read from buffer
2277  * @buffer: the buffer to allocate for.
2278  *
2279  * This function is used in conjunction with ring_buffer_read_page.
2280  * When reading a full page from the ring buffer, these functions
2281  * can be used to speed up the process. The calling function should
2282  * allocate a few pages first with this function. Then when it
2283  * needs to get pages from the ring buffer, it passes the result
2284  * of this function into ring_buffer_read_page, which will swap
2285  * the page that was allocated, with the read page of the buffer.
2286  *
2287  * Returns:
2288  *  The page allocated, or NULL on error.
2289  */
2290 void *ring_buffer_alloc_read_page(struct ring_buffer *buffer)
2291 {
2292         unsigned long addr;
2293         struct buffer_data_page *bpage;
2294
2295         addr = __get_free_page(GFP_KERNEL);
2296         if (!addr)
2297                 return NULL;
2298
2299         bpage = (void *)addr;
2300
2301         return bpage;
2302 }
2303
2304 /**
2305  * ring_buffer_free_read_page - free an allocated read page
2306  * @buffer: the buffer the page was allocate for
2307  * @data: the page to free
2308  *
2309  * Free a page allocated from ring_buffer_alloc_read_page.
2310  */
2311 void ring_buffer_free_read_page(struct ring_buffer *buffer, void *data)
2312 {
2313         free_page((unsigned long)data);
2314 }
2315
2316 /**
2317  * ring_buffer_read_page - extract a page from the ring buffer
2318  * @buffer: buffer to extract from
2319  * @data_page: the page to use allocated from ring_buffer_alloc_read_page
2320  * @cpu: the cpu of the buffer to extract
2321  * @full: should the extraction only happen when the page is full.
2322  *
2323  * This function will pull out a page from the ring buffer and consume it.
2324  * @data_page must be the address of the variable that was returned
2325  * from ring_buffer_alloc_read_page. This is because the page might be used
2326  * to swap with a page in the ring buffer.
2327  *
2328  * for example:
2329  *      rpage = ring_buffer_alloc_page(buffer);
2330  *      if (!rpage)
2331  *              return error;
2332  *      ret = ring_buffer_read_page(buffer, &rpage, cpu, 0);
2333  *      if (ret)
2334  *              process_page(rpage);
2335  *
2336  * When @full is set, the function will not return true unless
2337  * the writer is off the reader page.
2338  *
2339  * Note: it is up to the calling functions to handle sleeps and wakeups.
2340  *  The ring buffer can be used anywhere in the kernel and can not
2341  *  blindly call wake_up. The layer that uses the ring buffer must be
2342  *  responsible for that.
2343  *
2344  * Returns:
2345  *  1 if data has been transferred
2346  *  0 if no data has been transferred.
2347  */
2348 int ring_buffer_read_page(struct ring_buffer *buffer,
2349                             void **data_page, int cpu, int full)
2350 {
2351         struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
2352         struct ring_buffer_event *event;
2353         struct buffer_data_page *bpage;
2354         unsigned long flags;
2355         int ret = 0;
2356
2357         if (!data_page)
2358                 return 0;
2359
2360         bpage = *data_page;
2361         if (!bpage)
2362                 return 0;
2363
2364         spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
2365
2366         /*
2367          * rb_buffer_peek will get the next ring buffer if
2368          * the current reader page is empty.
2369          */
2370         event = rb_buffer_peek(buffer, cpu, NULL);
2371         if (!event)
2372                 goto out;
2373
2374         /* check for data */
2375         if (!local_read(&cpu_buffer->reader_page->page->commit))
2376                 goto out;
2377         /*
2378          * If the writer is already off of the read page, then simply
2379          * switch the read page with the given page. Otherwise
2380          * we need to copy the data from the reader to the writer.
2381          */
2382         if (cpu_buffer->reader_page == cpu_buffer->commit_page) {
2383                 unsigned int read = cpu_buffer->reader_page->read;
2384
2385                 if (full)
2386                         goto out;
2387                 /* The writer is still on the reader page, we must copy */
2388                 bpage = cpu_buffer->reader_page->page;
2389                 memcpy(bpage->data,
2390                        cpu_buffer->reader_page->page->data + read,
2391                        local_read(&bpage->commit) - read);
2392
2393                 /* consume what was read */
2394                 cpu_buffer->reader_page += read;
2395
2396         } else {
2397                 /* swap the pages */
2398                 rb_init_page(bpage);
2399                 bpage = cpu_buffer->reader_page->page;
2400                 cpu_buffer->reader_page->page = *data_page;
2401                 cpu_buffer->reader_page->read = 0;
2402                 *data_page = bpage;
2403         }
2404         ret = 1;
2405
2406         /* update the entry counter */
2407         rb_remove_entries(cpu_buffer, bpage);
2408  out:
2409         spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
2410
2411         return ret;
2412 }
2413
2414 static ssize_t
2415 rb_simple_read(struct file *filp, char __user *ubuf,
2416                size_t cnt, loff_t *ppos)
2417 {
2418         long *p = filp->private_data;
2419         char buf[64];
2420         int r;
2421
2422         if (test_bit(RB_BUFFERS_DISABLED_BIT, p))
2423                 r = sprintf(buf, "permanently disabled\n");
2424         else
2425                 r = sprintf(buf, "%d\n", test_bit(RB_BUFFERS_ON_BIT, p));
2426
2427         return simple_read_from_buffer(ubuf, cnt, ppos, buf, r);
2428 }
2429
2430 static ssize_t
2431 rb_simple_write(struct file *filp, const char __user *ubuf,
2432                 size_t cnt, loff_t *ppos)
2433 {
2434         long *p = filp->private_data;
2435         char buf[64];
2436         long val;
2437         int ret;
2438
2439         if (cnt >= sizeof(buf))
2440                 return -EINVAL;
2441
2442         if (copy_from_user(&buf, ubuf, cnt))
2443                 return -EFAULT;
2444
2445         buf[cnt] = 0;
2446
2447         ret = strict_strtoul(buf, 10, &val);
2448         if (ret < 0)
2449                 return ret;
2450
2451         if (val)
2452                 set_bit(RB_BUFFERS_ON_BIT, p);
2453         else
2454                 clear_bit(RB_BUFFERS_ON_BIT, p);
2455
2456         (*ppos)++;
2457
2458         return cnt;
2459 }
2460
2461 static struct file_operations rb_simple_fops = {
2462         .open           = tracing_open_generic,
2463         .read           = rb_simple_read,
2464         .write          = rb_simple_write,
2465 };
2466
2467
2468 static __init int rb_init_debugfs(void)
2469 {
2470         struct dentry *d_tracer;
2471         struct dentry *entry;
2472
2473         d_tracer = tracing_init_dentry();
2474
2475         entry = debugfs_create_file("tracing_on", 0644, d_tracer,
2476                                     &ring_buffer_flags, &rb_simple_fops);
2477         if (!entry)
2478                 pr_warning("Could not create debugfs 'tracing_on' entry\n");
2479
2480         return 0;
2481 }
2482
2483 fs_initcall(rb_init_debugfs);