4 * Copyright (C) 2008 Steven Rostedt <srostedt@redhat.com>
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>
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.
28 * There's three layers that must be on in order to write
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.
35 * In case of an anomaly, this global flag has a bit set that
36 * will permantly disable all ring buffers.
40 * Global flag to disable all recording to ring buffers
41 * This has two bits: ON, DISABLED
45 * 0 0 : ring buffers are off
46 * 1 0 : ring buffers are on
47 * X 1 : ring buffers are permanently disabled
51 RB_BUFFERS_ON_BIT = 0,
52 RB_BUFFERS_DISABLED_BIT = 1,
56 RB_BUFFERS_ON = 1 << RB_BUFFERS_ON_BIT,
57 RB_BUFFERS_DISABLED = 1 << RB_BUFFERS_DISABLED_BIT,
60 static long ring_buffer_flags __read_mostly = RB_BUFFERS_ON;
63 * tracing_on - enable all tracing buffers
65 * This function enables all tracing buffers that may have been
66 * disabled with tracing_off.
70 set_bit(RB_BUFFERS_ON_BIT, &ring_buffer_flags);
72 EXPORT_SYMBOL_GPL(tracing_on);
75 * tracing_off - turn off all tracing buffers
77 * This function stops all tracing buffers from recording data.
78 * It does not disable any overhead the tracers themselves may
79 * be causing. This function simply causes all recording to
80 * the ring buffers to fail.
82 void tracing_off(void)
84 clear_bit(RB_BUFFERS_ON_BIT, &ring_buffer_flags);
86 EXPORT_SYMBOL_GPL(tracing_off);
89 * tracing_off_permanent - permanently disable ring buffers
91 * This function, once called, will disable all ring buffers
94 void tracing_off_permanent(void)
96 set_bit(RB_BUFFERS_DISABLED_BIT, &ring_buffer_flags);
101 /* Up this if you want to test the TIME_EXTENTS and normalization */
102 #define DEBUG_SHIFT 0
105 u64 ring_buffer_time_stamp(int cpu)
109 preempt_disable_notrace();
110 /* shift to debug/test normalization and TIME_EXTENTS */
111 time = sched_clock() << DEBUG_SHIFT;
112 preempt_enable_no_resched_notrace();
116 EXPORT_SYMBOL_GPL(ring_buffer_time_stamp);
118 void ring_buffer_normalize_time_stamp(int cpu, u64 *ts)
120 /* Just stupid testing the normalize function and deltas */
123 EXPORT_SYMBOL_GPL(ring_buffer_normalize_time_stamp);
125 #define RB_EVNT_HDR_SIZE (sizeof(struct ring_buffer_event))
126 #define RB_ALIGNMENT_SHIFT 2
127 #define RB_ALIGNMENT (1 << RB_ALIGNMENT_SHIFT)
128 #define RB_MAX_SMALL_DATA 28
131 RB_LEN_TIME_EXTEND = 8,
132 RB_LEN_TIME_STAMP = 16,
135 /* inline for ring buffer fast paths */
137 rb_event_length(struct ring_buffer_event *event)
141 switch (event->type) {
142 case RINGBUF_TYPE_PADDING:
146 case RINGBUF_TYPE_TIME_EXTEND:
147 return RB_LEN_TIME_EXTEND;
149 case RINGBUF_TYPE_TIME_STAMP:
150 return RB_LEN_TIME_STAMP;
152 case RINGBUF_TYPE_DATA:
154 length = event->len << RB_ALIGNMENT_SHIFT;
156 length = event->array[0];
157 return length + RB_EVNT_HDR_SIZE;
166 * ring_buffer_event_length - return the length of the event
167 * @event: the event to get the length of
169 unsigned ring_buffer_event_length(struct ring_buffer_event *event)
171 unsigned length = rb_event_length(event);
172 if (event->type != RINGBUF_TYPE_DATA)
174 length -= RB_EVNT_HDR_SIZE;
175 if (length > RB_MAX_SMALL_DATA + sizeof(event->array[0]))
176 length -= sizeof(event->array[0]);
179 EXPORT_SYMBOL_GPL(ring_buffer_event_length);
181 /* inline for ring buffer fast paths */
183 rb_event_data(struct ring_buffer_event *event)
185 BUG_ON(event->type != RINGBUF_TYPE_DATA);
186 /* If length is in len field, then array[0] has the data */
188 return (void *)&event->array[0];
189 /* Otherwise length is in array[0] and array[1] has the data */
190 return (void *)&event->array[1];
194 * ring_buffer_event_data - return the data of the event
195 * @event: the event to get the data from
197 void *ring_buffer_event_data(struct ring_buffer_event *event)
199 return rb_event_data(event);
201 EXPORT_SYMBOL_GPL(ring_buffer_event_data);
203 #define for_each_buffer_cpu(buffer, cpu) \
204 for_each_cpu(cpu, buffer->cpumask)
207 #define TS_MASK ((1ULL << TS_SHIFT) - 1)
208 #define TS_DELTA_TEST (~TS_MASK)
210 struct buffer_data_page {
211 u64 time_stamp; /* page time stamp */
212 local_t commit; /* write commited index */
213 unsigned char data[]; /* data of buffer page */
217 local_t write; /* index for next write */
218 unsigned read; /* index for next read */
219 struct list_head list; /* list of free pages */
220 struct buffer_data_page *page; /* Actual data page */
223 static void rb_init_page(struct buffer_data_page *bpage)
225 local_set(&bpage->commit, 0);
229 * Also stolen from mm/slob.c. Thanks to Mathieu Desnoyers for pointing
232 static void free_buffer_page(struct buffer_page *bpage)
234 free_page((unsigned long)bpage->page);
239 * We need to fit the time_stamp delta into 27 bits.
241 static inline int test_time_stamp(u64 delta)
243 if (delta & TS_DELTA_TEST)
248 #define BUF_PAGE_SIZE (PAGE_SIZE - sizeof(struct buffer_data_page))
251 * head_page == tail_page && head == tail then buffer is empty.
253 struct ring_buffer_per_cpu {
255 struct ring_buffer *buffer;
256 spinlock_t reader_lock; /* serialize readers */
258 struct lock_class_key lock_key;
259 struct list_head pages;
260 struct buffer_page *head_page; /* read from head */
261 struct buffer_page *tail_page; /* write to tail */
262 struct buffer_page *commit_page; /* commited pages */
263 struct buffer_page *reader_page;
264 unsigned long overrun;
265 unsigned long entries;
268 atomic_t record_disabled;
275 cpumask_var_t cpumask;
276 atomic_t record_disabled;
280 struct ring_buffer_per_cpu **buffers;
283 struct ring_buffer_iter {
284 struct ring_buffer_per_cpu *cpu_buffer;
286 struct buffer_page *head_page;
290 /* buffer may be either ring_buffer or ring_buffer_per_cpu */
291 #define RB_WARN_ON(buffer, cond) \
293 int _____ret = unlikely(cond); \
295 atomic_inc(&buffer->record_disabled); \
302 * check_pages - integrity check of buffer pages
303 * @cpu_buffer: CPU buffer with pages to test
305 * As a safty measure we check to make sure the data pages have not
308 static int rb_check_pages(struct ring_buffer_per_cpu *cpu_buffer)
310 struct list_head *head = &cpu_buffer->pages;
311 struct buffer_page *bpage, *tmp;
313 if (RB_WARN_ON(cpu_buffer, head->next->prev != head))
315 if (RB_WARN_ON(cpu_buffer, head->prev->next != head))
318 list_for_each_entry_safe(bpage, tmp, head, list) {
319 if (RB_WARN_ON(cpu_buffer,
320 bpage->list.next->prev != &bpage->list))
322 if (RB_WARN_ON(cpu_buffer,
323 bpage->list.prev->next != &bpage->list))
330 static int rb_allocate_pages(struct ring_buffer_per_cpu *cpu_buffer,
333 struct list_head *head = &cpu_buffer->pages;
334 struct buffer_page *bpage, *tmp;
339 for (i = 0; i < nr_pages; i++) {
340 bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),
341 GFP_KERNEL, cpu_to_node(cpu_buffer->cpu));
344 list_add(&bpage->list, &pages);
346 addr = __get_free_page(GFP_KERNEL);
349 bpage->page = (void *)addr;
350 rb_init_page(bpage->page);
353 list_splice(&pages, head);
355 rb_check_pages(cpu_buffer);
360 list_for_each_entry_safe(bpage, tmp, &pages, list) {
361 list_del_init(&bpage->list);
362 free_buffer_page(bpage);
367 static struct ring_buffer_per_cpu *
368 rb_allocate_cpu_buffer(struct ring_buffer *buffer, int cpu)
370 struct ring_buffer_per_cpu *cpu_buffer;
371 struct buffer_page *bpage;
375 cpu_buffer = kzalloc_node(ALIGN(sizeof(*cpu_buffer), cache_line_size()),
376 GFP_KERNEL, cpu_to_node(cpu));
380 cpu_buffer->cpu = cpu;
381 cpu_buffer->buffer = buffer;
382 spin_lock_init(&cpu_buffer->reader_lock);
383 cpu_buffer->lock = (raw_spinlock_t)__RAW_SPIN_LOCK_UNLOCKED;
384 INIT_LIST_HEAD(&cpu_buffer->pages);
386 bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),
387 GFP_KERNEL, cpu_to_node(cpu));
389 goto fail_free_buffer;
391 cpu_buffer->reader_page = bpage;
392 addr = __get_free_page(GFP_KERNEL);
394 goto fail_free_reader;
395 bpage->page = (void *)addr;
396 rb_init_page(bpage->page);
398 INIT_LIST_HEAD(&cpu_buffer->reader_page->list);
400 ret = rb_allocate_pages(cpu_buffer, buffer->pages);
402 goto fail_free_reader;
404 cpu_buffer->head_page
405 = list_entry(cpu_buffer->pages.next, struct buffer_page, list);
406 cpu_buffer->tail_page = cpu_buffer->commit_page = cpu_buffer->head_page;
411 free_buffer_page(cpu_buffer->reader_page);
418 static void rb_free_cpu_buffer(struct ring_buffer_per_cpu *cpu_buffer)
420 struct list_head *head = &cpu_buffer->pages;
421 struct buffer_page *bpage, *tmp;
423 list_del_init(&cpu_buffer->reader_page->list);
424 free_buffer_page(cpu_buffer->reader_page);
426 list_for_each_entry_safe(bpage, tmp, head, list) {
427 list_del_init(&bpage->list);
428 free_buffer_page(bpage);
434 * Causes compile errors if the struct buffer_page gets bigger
435 * than the struct page.
437 extern int ring_buffer_page_too_big(void);
440 * ring_buffer_alloc - allocate a new ring_buffer
441 * @size: the size in bytes per cpu that is needed.
442 * @flags: attributes to set for the ring buffer.
444 * Currently the only flag that is available is the RB_FL_OVERWRITE
445 * flag. This flag means that the buffer will overwrite old data
446 * when the buffer wraps. If this flag is not set, the buffer will
447 * drop data when the tail hits the head.
449 struct ring_buffer *ring_buffer_alloc(unsigned long size, unsigned flags)
451 struct ring_buffer *buffer;
455 /* Paranoid! Optimizes out when all is well */
456 if (sizeof(struct buffer_page) > sizeof(struct page))
457 ring_buffer_page_too_big();
460 /* keep it in its own cache line */
461 buffer = kzalloc(ALIGN(sizeof(*buffer), cache_line_size()),
466 if (!alloc_cpumask_var(&buffer->cpumask, GFP_KERNEL))
467 goto fail_free_buffer;
469 buffer->pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
470 buffer->flags = flags;
472 /* need at least two pages */
473 if (buffer->pages == 1)
476 cpumask_copy(buffer->cpumask, cpu_possible_mask);
477 buffer->cpus = nr_cpu_ids;
479 bsize = sizeof(void *) * nr_cpu_ids;
480 buffer->buffers = kzalloc(ALIGN(bsize, cache_line_size()),
482 if (!buffer->buffers)
483 goto fail_free_cpumask;
485 for_each_buffer_cpu(buffer, cpu) {
486 buffer->buffers[cpu] =
487 rb_allocate_cpu_buffer(buffer, cpu);
488 if (!buffer->buffers[cpu])
489 goto fail_free_buffers;
492 mutex_init(&buffer->mutex);
497 for_each_buffer_cpu(buffer, cpu) {
498 if (buffer->buffers[cpu])
499 rb_free_cpu_buffer(buffer->buffers[cpu]);
501 kfree(buffer->buffers);
504 free_cpumask_var(buffer->cpumask);
510 EXPORT_SYMBOL_GPL(ring_buffer_alloc);
513 * ring_buffer_free - free a ring buffer.
514 * @buffer: the buffer to free.
517 ring_buffer_free(struct ring_buffer *buffer)
521 for_each_buffer_cpu(buffer, cpu)
522 rb_free_cpu_buffer(buffer->buffers[cpu]);
524 free_cpumask_var(buffer->cpumask);
528 EXPORT_SYMBOL_GPL(ring_buffer_free);
530 static void rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer);
533 rb_remove_pages(struct ring_buffer_per_cpu *cpu_buffer, unsigned nr_pages)
535 struct buffer_page *bpage;
539 atomic_inc(&cpu_buffer->record_disabled);
542 for (i = 0; i < nr_pages; i++) {
543 if (RB_WARN_ON(cpu_buffer, list_empty(&cpu_buffer->pages)))
545 p = cpu_buffer->pages.next;
546 bpage = list_entry(p, struct buffer_page, list);
547 list_del_init(&bpage->list);
548 free_buffer_page(bpage);
550 if (RB_WARN_ON(cpu_buffer, list_empty(&cpu_buffer->pages)))
553 rb_reset_cpu(cpu_buffer);
555 rb_check_pages(cpu_buffer);
557 atomic_dec(&cpu_buffer->record_disabled);
562 rb_insert_pages(struct ring_buffer_per_cpu *cpu_buffer,
563 struct list_head *pages, unsigned nr_pages)
565 struct buffer_page *bpage;
569 atomic_inc(&cpu_buffer->record_disabled);
572 for (i = 0; i < nr_pages; i++) {
573 if (RB_WARN_ON(cpu_buffer, list_empty(pages)))
576 bpage = list_entry(p, struct buffer_page, list);
577 list_del_init(&bpage->list);
578 list_add_tail(&bpage->list, &cpu_buffer->pages);
580 rb_reset_cpu(cpu_buffer);
582 rb_check_pages(cpu_buffer);
584 atomic_dec(&cpu_buffer->record_disabled);
588 * ring_buffer_resize - resize the ring buffer
589 * @buffer: the buffer to resize.
590 * @size: the new size.
592 * The tracer is responsible for making sure that the buffer is
593 * not being used while changing the size.
594 * Note: We may be able to change the above requirement by using
595 * RCU synchronizations.
597 * Minimum size is 2 * BUF_PAGE_SIZE.
599 * Returns -1 on failure.
601 int ring_buffer_resize(struct ring_buffer *buffer, unsigned long size)
603 struct ring_buffer_per_cpu *cpu_buffer;
604 unsigned nr_pages, rm_pages, new_pages;
605 struct buffer_page *bpage, *tmp;
606 unsigned long buffer_size;
612 * Always succeed at resizing a non-existent buffer:
617 size = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
618 size *= BUF_PAGE_SIZE;
619 buffer_size = buffer->pages * BUF_PAGE_SIZE;
621 /* we need a minimum of two pages */
622 if (size < BUF_PAGE_SIZE * 2)
623 size = BUF_PAGE_SIZE * 2;
625 if (size == buffer_size)
628 mutex_lock(&buffer->mutex);
630 nr_pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
632 if (size < buffer_size) {
634 /* easy case, just free pages */
635 if (RB_WARN_ON(buffer, nr_pages >= buffer->pages)) {
636 mutex_unlock(&buffer->mutex);
640 rm_pages = buffer->pages - nr_pages;
642 for_each_buffer_cpu(buffer, cpu) {
643 cpu_buffer = buffer->buffers[cpu];
644 rb_remove_pages(cpu_buffer, rm_pages);
650 * This is a bit more difficult. We only want to add pages
651 * when we can allocate enough for all CPUs. We do this
652 * by allocating all the pages and storing them on a local
653 * link list. If we succeed in our allocation, then we
654 * add these pages to the cpu_buffers. Otherwise we just free
655 * them all and return -ENOMEM;
657 if (RB_WARN_ON(buffer, nr_pages <= buffer->pages)) {
658 mutex_unlock(&buffer->mutex);
662 new_pages = nr_pages - buffer->pages;
664 for_each_buffer_cpu(buffer, cpu) {
665 for (i = 0; i < new_pages; i++) {
666 bpage = kzalloc_node(ALIGN(sizeof(*bpage),
668 GFP_KERNEL, cpu_to_node(cpu));
671 list_add(&bpage->list, &pages);
672 addr = __get_free_page(GFP_KERNEL);
675 bpage->page = (void *)addr;
676 rb_init_page(bpage->page);
680 for_each_buffer_cpu(buffer, cpu) {
681 cpu_buffer = buffer->buffers[cpu];
682 rb_insert_pages(cpu_buffer, &pages, new_pages);
685 if (RB_WARN_ON(buffer, !list_empty(&pages))) {
686 mutex_unlock(&buffer->mutex);
691 buffer->pages = nr_pages;
692 mutex_unlock(&buffer->mutex);
697 list_for_each_entry_safe(bpage, tmp, &pages, list) {
698 list_del_init(&bpage->list);
699 free_buffer_page(bpage);
701 mutex_unlock(&buffer->mutex);
704 EXPORT_SYMBOL_GPL(ring_buffer_resize);
706 static inline int rb_null_event(struct ring_buffer_event *event)
708 return event->type == RINGBUF_TYPE_PADDING;
712 __rb_data_page_index(struct buffer_data_page *bpage, unsigned index)
714 return bpage->data + index;
717 static inline void *__rb_page_index(struct buffer_page *bpage, unsigned index)
719 return bpage->page->data + index;
722 static inline struct ring_buffer_event *
723 rb_reader_event(struct ring_buffer_per_cpu *cpu_buffer)
725 return __rb_page_index(cpu_buffer->reader_page,
726 cpu_buffer->reader_page->read);
729 static inline struct ring_buffer_event *
730 rb_head_event(struct ring_buffer_per_cpu *cpu_buffer)
732 return __rb_page_index(cpu_buffer->head_page,
733 cpu_buffer->head_page->read);
736 static inline struct ring_buffer_event *
737 rb_iter_head_event(struct ring_buffer_iter *iter)
739 return __rb_page_index(iter->head_page, iter->head);
742 static inline unsigned rb_page_write(struct buffer_page *bpage)
744 return local_read(&bpage->write);
747 static inline unsigned rb_page_commit(struct buffer_page *bpage)
749 return local_read(&bpage->page->commit);
752 /* Size is determined by what has been commited */
753 static inline unsigned rb_page_size(struct buffer_page *bpage)
755 return rb_page_commit(bpage);
758 static inline unsigned
759 rb_commit_index(struct ring_buffer_per_cpu *cpu_buffer)
761 return rb_page_commit(cpu_buffer->commit_page);
764 static inline unsigned rb_head_size(struct ring_buffer_per_cpu *cpu_buffer)
766 return rb_page_commit(cpu_buffer->head_page);
770 * When the tail hits the head and the buffer is in overwrite mode,
771 * the head jumps to the next page and all content on the previous
772 * page is discarded. But before doing so, we update the overrun
773 * variable of the buffer.
775 static void rb_update_overflow(struct ring_buffer_per_cpu *cpu_buffer)
777 struct ring_buffer_event *event;
780 for (head = 0; head < rb_head_size(cpu_buffer);
781 head += rb_event_length(event)) {
783 event = __rb_page_index(cpu_buffer->head_page, head);
784 if (RB_WARN_ON(cpu_buffer, rb_null_event(event)))
786 /* Only count data entries */
787 if (event->type != RINGBUF_TYPE_DATA)
789 cpu_buffer->overrun++;
790 cpu_buffer->entries--;
794 static inline void rb_inc_page(struct ring_buffer_per_cpu *cpu_buffer,
795 struct buffer_page **bpage)
797 struct list_head *p = (*bpage)->list.next;
799 if (p == &cpu_buffer->pages)
802 *bpage = list_entry(p, struct buffer_page, list);
805 static inline unsigned
806 rb_event_index(struct ring_buffer_event *event)
808 unsigned long addr = (unsigned long)event;
810 return (addr & ~PAGE_MASK) - (PAGE_SIZE - BUF_PAGE_SIZE);
814 rb_is_commit(struct ring_buffer_per_cpu *cpu_buffer,
815 struct ring_buffer_event *event)
817 unsigned long addr = (unsigned long)event;
820 index = rb_event_index(event);
823 return cpu_buffer->commit_page->page == (void *)addr &&
824 rb_commit_index(cpu_buffer) == index;
828 rb_set_commit_event(struct ring_buffer_per_cpu *cpu_buffer,
829 struct ring_buffer_event *event)
831 unsigned long addr = (unsigned long)event;
834 index = rb_event_index(event);
837 while (cpu_buffer->commit_page->page != (void *)addr) {
838 if (RB_WARN_ON(cpu_buffer,
839 cpu_buffer->commit_page == cpu_buffer->tail_page))
841 cpu_buffer->commit_page->page->commit =
842 cpu_buffer->commit_page->write;
843 rb_inc_page(cpu_buffer, &cpu_buffer->commit_page);
844 cpu_buffer->write_stamp =
845 cpu_buffer->commit_page->page->time_stamp;
848 /* Now set the commit to the event's index */
849 local_set(&cpu_buffer->commit_page->page->commit, index);
853 rb_set_commit_to_write(struct ring_buffer_per_cpu *cpu_buffer)
856 * We only race with interrupts and NMIs on this CPU.
857 * If we own the commit event, then we can commit
858 * all others that interrupted us, since the interruptions
859 * are in stack format (they finish before they come
860 * back to us). This allows us to do a simple loop to
861 * assign the commit to the tail.
864 while (cpu_buffer->commit_page != cpu_buffer->tail_page) {
865 cpu_buffer->commit_page->page->commit =
866 cpu_buffer->commit_page->write;
867 rb_inc_page(cpu_buffer, &cpu_buffer->commit_page);
868 cpu_buffer->write_stamp =
869 cpu_buffer->commit_page->page->time_stamp;
870 /* add barrier to keep gcc from optimizing too much */
873 while (rb_commit_index(cpu_buffer) !=
874 rb_page_write(cpu_buffer->commit_page)) {
875 cpu_buffer->commit_page->page->commit =
876 cpu_buffer->commit_page->write;
880 /* again, keep gcc from optimizing */
884 * If an interrupt came in just after the first while loop
885 * and pushed the tail page forward, we will be left with
886 * a dangling commit that will never go forward.
888 if (unlikely(cpu_buffer->commit_page != cpu_buffer->tail_page))
892 static void rb_reset_reader_page(struct ring_buffer_per_cpu *cpu_buffer)
894 cpu_buffer->read_stamp = cpu_buffer->reader_page->page->time_stamp;
895 cpu_buffer->reader_page->read = 0;
898 static void rb_inc_iter(struct ring_buffer_iter *iter)
900 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
903 * The iterator could be on the reader page (it starts there).
904 * But the head could have moved, since the reader was
905 * found. Check for this case and assign the iterator
906 * to the head page instead of next.
908 if (iter->head_page == cpu_buffer->reader_page)
909 iter->head_page = cpu_buffer->head_page;
911 rb_inc_page(cpu_buffer, &iter->head_page);
913 iter->read_stamp = iter->head_page->page->time_stamp;
918 * ring_buffer_update_event - update event type and data
919 * @event: the even to update
920 * @type: the type of event
921 * @length: the size of the event field in the ring buffer
923 * Update the type and data fields of the event. The length
924 * is the actual size that is written to the ring buffer,
925 * and with this, we can determine what to place into the
929 rb_update_event(struct ring_buffer_event *event,
930 unsigned type, unsigned length)
936 case RINGBUF_TYPE_PADDING:
939 case RINGBUF_TYPE_TIME_EXTEND:
941 (RB_LEN_TIME_EXTEND + (RB_ALIGNMENT-1))
942 >> RB_ALIGNMENT_SHIFT;
945 case RINGBUF_TYPE_TIME_STAMP:
947 (RB_LEN_TIME_STAMP + (RB_ALIGNMENT-1))
948 >> RB_ALIGNMENT_SHIFT;
951 case RINGBUF_TYPE_DATA:
952 length -= RB_EVNT_HDR_SIZE;
953 if (length > RB_MAX_SMALL_DATA) {
955 event->array[0] = length;
958 (length + (RB_ALIGNMENT-1))
959 >> RB_ALIGNMENT_SHIFT;
966 static unsigned rb_calculate_event_length(unsigned length)
968 struct ring_buffer_event event; /* Used only for sizeof array */
970 /* zero length can cause confusions */
974 if (length > RB_MAX_SMALL_DATA)
975 length += sizeof(event.array[0]);
977 length += RB_EVNT_HDR_SIZE;
978 length = ALIGN(length, RB_ALIGNMENT);
983 static struct ring_buffer_event *
984 __rb_reserve_next(struct ring_buffer_per_cpu *cpu_buffer,
985 unsigned type, unsigned long length, u64 *ts)
987 struct buffer_page *tail_page, *head_page, *reader_page, *commit_page;
988 unsigned long tail, write;
989 struct ring_buffer *buffer = cpu_buffer->buffer;
990 struct ring_buffer_event *event;
993 commit_page = cpu_buffer->commit_page;
994 /* we just need to protect against interrupts */
996 tail_page = cpu_buffer->tail_page;
997 write = local_add_return(length, &tail_page->write);
998 tail = write - length;
1000 /* See if we shot pass the end of this buffer page */
1001 if (write > BUF_PAGE_SIZE) {
1002 struct buffer_page *next_page = tail_page;
1004 local_irq_save(flags);
1005 __raw_spin_lock(&cpu_buffer->lock);
1007 rb_inc_page(cpu_buffer, &next_page);
1009 head_page = cpu_buffer->head_page;
1010 reader_page = cpu_buffer->reader_page;
1012 /* we grabbed the lock before incrementing */
1013 if (RB_WARN_ON(cpu_buffer, next_page == reader_page))
1017 * If for some reason, we had an interrupt storm that made
1018 * it all the way around the buffer, bail, and warn
1021 if (unlikely(next_page == commit_page)) {
1026 if (next_page == head_page) {
1027 if (!(buffer->flags & RB_FL_OVERWRITE)) {
1029 if (tail <= BUF_PAGE_SIZE)
1030 local_set(&tail_page->write, tail);
1034 /* tail_page has not moved yet? */
1035 if (tail_page == cpu_buffer->tail_page) {
1036 /* count overflows */
1037 rb_update_overflow(cpu_buffer);
1039 rb_inc_page(cpu_buffer, &head_page);
1040 cpu_buffer->head_page = head_page;
1041 cpu_buffer->head_page->read = 0;
1046 * If the tail page is still the same as what we think
1047 * it is, then it is up to us to update the tail
1050 if (tail_page == cpu_buffer->tail_page) {
1051 local_set(&next_page->write, 0);
1052 local_set(&next_page->page->commit, 0);
1053 cpu_buffer->tail_page = next_page;
1055 /* reread the time stamp */
1056 *ts = ring_buffer_time_stamp(cpu_buffer->cpu);
1057 cpu_buffer->tail_page->page->time_stamp = *ts;
1061 * The actual tail page has moved forward.
1063 if (tail < BUF_PAGE_SIZE) {
1064 /* Mark the rest of the page with padding */
1065 event = __rb_page_index(tail_page, tail);
1066 event->type = RINGBUF_TYPE_PADDING;
1069 if (tail <= BUF_PAGE_SIZE)
1070 /* Set the write back to the previous setting */
1071 local_set(&tail_page->write, tail);
1074 * If this was a commit entry that failed,
1075 * increment that too
1077 if (tail_page == cpu_buffer->commit_page &&
1078 tail == rb_commit_index(cpu_buffer)) {
1079 rb_set_commit_to_write(cpu_buffer);
1082 __raw_spin_unlock(&cpu_buffer->lock);
1083 local_irq_restore(flags);
1085 /* fail and let the caller try again */
1086 return ERR_PTR(-EAGAIN);
1089 /* We reserved something on the buffer */
1091 if (RB_WARN_ON(cpu_buffer, write > BUF_PAGE_SIZE))
1094 event = __rb_page_index(tail_page, tail);
1095 rb_update_event(event, type, length);
1098 * If this is a commit and the tail is zero, then update
1099 * this page's time stamp.
1101 if (!tail && rb_is_commit(cpu_buffer, event))
1102 cpu_buffer->commit_page->page->time_stamp = *ts;
1107 __raw_spin_unlock(&cpu_buffer->lock);
1108 local_irq_restore(flags);
1113 rb_add_time_stamp(struct ring_buffer_per_cpu *cpu_buffer,
1114 u64 *ts, u64 *delta)
1116 struct ring_buffer_event *event;
1120 if (unlikely(*delta > (1ULL << 59) && !once++)) {
1121 printk(KERN_WARNING "Delta way too big! %llu"
1122 " ts=%llu write stamp = %llu\n",
1123 (unsigned long long)*delta,
1124 (unsigned long long)*ts,
1125 (unsigned long long)cpu_buffer->write_stamp);
1130 * The delta is too big, we to add a
1133 event = __rb_reserve_next(cpu_buffer,
1134 RINGBUF_TYPE_TIME_EXTEND,
1140 if (PTR_ERR(event) == -EAGAIN)
1143 /* Only a commited time event can update the write stamp */
1144 if (rb_is_commit(cpu_buffer, event)) {
1146 * If this is the first on the page, then we need to
1147 * update the page itself, and just put in a zero.
1149 if (rb_event_index(event)) {
1150 event->time_delta = *delta & TS_MASK;
1151 event->array[0] = *delta >> TS_SHIFT;
1153 cpu_buffer->commit_page->page->time_stamp = *ts;
1154 event->time_delta = 0;
1155 event->array[0] = 0;
1157 cpu_buffer->write_stamp = *ts;
1158 /* let the caller know this was the commit */
1161 /* Darn, this is just wasted space */
1162 event->time_delta = 0;
1163 event->array[0] = 0;
1172 static struct ring_buffer_event *
1173 rb_reserve_next_event(struct ring_buffer_per_cpu *cpu_buffer,
1174 unsigned type, unsigned long length)
1176 struct ring_buffer_event *event;
1183 * We allow for interrupts to reenter here and do a trace.
1184 * If one does, it will cause this original code to loop
1185 * back here. Even with heavy interrupts happening, this
1186 * should only happen a few times in a row. If this happens
1187 * 1000 times in a row, there must be either an interrupt
1188 * storm or we have something buggy.
1191 if (RB_WARN_ON(cpu_buffer, ++nr_loops > 1000))
1194 ts = ring_buffer_time_stamp(cpu_buffer->cpu);
1197 * Only the first commit can update the timestamp.
1198 * Yes there is a race here. If an interrupt comes in
1199 * just after the conditional and it traces too, then it
1200 * will also check the deltas. More than one timestamp may
1201 * also be made. But only the entry that did the actual
1202 * commit will be something other than zero.
1204 if (cpu_buffer->tail_page == cpu_buffer->commit_page &&
1205 rb_page_write(cpu_buffer->tail_page) ==
1206 rb_commit_index(cpu_buffer)) {
1208 delta = ts - cpu_buffer->write_stamp;
1210 /* make sure this delta is calculated here */
1213 /* Did the write stamp get updated already? */
1214 if (unlikely(ts < cpu_buffer->write_stamp))
1217 if (test_time_stamp(delta)) {
1219 commit = rb_add_time_stamp(cpu_buffer, &ts, &delta);
1221 if (commit == -EBUSY)
1224 if (commit == -EAGAIN)
1227 RB_WARN_ON(cpu_buffer, commit < 0);
1230 /* Non commits have zero deltas */
1233 event = __rb_reserve_next(cpu_buffer, type, length, &ts);
1234 if (PTR_ERR(event) == -EAGAIN)
1238 if (unlikely(commit))
1240 * Ouch! We needed a timestamp and it was commited. But
1241 * we didn't get our event reserved.
1243 rb_set_commit_to_write(cpu_buffer);
1248 * If the timestamp was commited, make the commit our entry
1249 * now so that we will update it when needed.
1252 rb_set_commit_event(cpu_buffer, event);
1253 else if (!rb_is_commit(cpu_buffer, event))
1256 event->time_delta = delta;
1261 static DEFINE_PER_CPU(int, rb_need_resched);
1264 * ring_buffer_lock_reserve - reserve a part of the buffer
1265 * @buffer: the ring buffer to reserve from
1266 * @length: the length of the data to reserve (excluding event header)
1267 * @flags: a pointer to save the interrupt flags
1269 * Returns a reseverd event on the ring buffer to copy directly to.
1270 * The user of this interface will need to get the body to write into
1271 * and can use the ring_buffer_event_data() interface.
1273 * The length is the length of the data needed, not the event length
1274 * which also includes the event header.
1276 * Must be paired with ring_buffer_unlock_commit, unless NULL is returned.
1277 * If NULL is returned, then nothing has been allocated or locked.
1279 struct ring_buffer_event *
1280 ring_buffer_lock_reserve(struct ring_buffer *buffer,
1281 unsigned long length,
1282 unsigned long *flags)
1284 struct ring_buffer_per_cpu *cpu_buffer;
1285 struct ring_buffer_event *event;
1288 if (ring_buffer_flags != RB_BUFFERS_ON)
1291 if (atomic_read(&buffer->record_disabled))
1294 /* If we are tracing schedule, we don't want to recurse */
1295 resched = ftrace_preempt_disable();
1297 cpu = raw_smp_processor_id();
1299 if (!cpumask_test_cpu(cpu, buffer->cpumask))
1302 cpu_buffer = buffer->buffers[cpu];
1304 if (atomic_read(&cpu_buffer->record_disabled))
1307 length = rb_calculate_event_length(length);
1308 if (length > BUF_PAGE_SIZE)
1311 event = rb_reserve_next_event(cpu_buffer, RINGBUF_TYPE_DATA, length);
1316 * Need to store resched state on this cpu.
1317 * Only the first needs to.
1320 if (preempt_count() == 1)
1321 per_cpu(rb_need_resched, cpu) = resched;
1326 ftrace_preempt_enable(resched);
1329 EXPORT_SYMBOL_GPL(ring_buffer_lock_reserve);
1331 static void rb_commit(struct ring_buffer_per_cpu *cpu_buffer,
1332 struct ring_buffer_event *event)
1334 cpu_buffer->entries++;
1336 /* Only process further if we own the commit */
1337 if (!rb_is_commit(cpu_buffer, event))
1340 cpu_buffer->write_stamp += event->time_delta;
1342 rb_set_commit_to_write(cpu_buffer);
1346 * ring_buffer_unlock_commit - commit a reserved
1347 * @buffer: The buffer to commit to
1348 * @event: The event pointer to commit.
1349 * @flags: the interrupt flags received from ring_buffer_lock_reserve.
1351 * This commits the data to the ring buffer, and releases any locks held.
1353 * Must be paired with ring_buffer_lock_reserve.
1355 int ring_buffer_unlock_commit(struct ring_buffer *buffer,
1356 struct ring_buffer_event *event,
1357 unsigned long flags)
1359 struct ring_buffer_per_cpu *cpu_buffer;
1360 int cpu = raw_smp_processor_id();
1362 cpu_buffer = buffer->buffers[cpu];
1364 rb_commit(cpu_buffer, event);
1367 * Only the last preempt count needs to restore preemption.
1369 if (preempt_count() == 1)
1370 ftrace_preempt_enable(per_cpu(rb_need_resched, cpu));
1372 preempt_enable_no_resched_notrace();
1376 EXPORT_SYMBOL_GPL(ring_buffer_unlock_commit);
1379 * ring_buffer_write - write data to the buffer without reserving
1380 * @buffer: The ring buffer to write to.
1381 * @length: The length of the data being written (excluding the event header)
1382 * @data: The data to write to the buffer.
1384 * This is like ring_buffer_lock_reserve and ring_buffer_unlock_commit as
1385 * one function. If you already have the data to write to the buffer, it
1386 * may be easier to simply call this function.
1388 * Note, like ring_buffer_lock_reserve, the length is the length of the data
1389 * and not the length of the event which would hold the header.
1391 int ring_buffer_write(struct ring_buffer *buffer,
1392 unsigned long length,
1395 struct ring_buffer_per_cpu *cpu_buffer;
1396 struct ring_buffer_event *event;
1397 unsigned long event_length;
1402 if (ring_buffer_flags != RB_BUFFERS_ON)
1405 if (atomic_read(&buffer->record_disabled))
1408 resched = ftrace_preempt_disable();
1410 cpu = raw_smp_processor_id();
1412 if (!cpumask_test_cpu(cpu, buffer->cpumask))
1415 cpu_buffer = buffer->buffers[cpu];
1417 if (atomic_read(&cpu_buffer->record_disabled))
1420 event_length = rb_calculate_event_length(length);
1421 event = rb_reserve_next_event(cpu_buffer,
1422 RINGBUF_TYPE_DATA, event_length);
1426 body = rb_event_data(event);
1428 memcpy(body, data, length);
1430 rb_commit(cpu_buffer, event);
1434 ftrace_preempt_enable(resched);
1438 EXPORT_SYMBOL_GPL(ring_buffer_write);
1440 static int rb_per_cpu_empty(struct ring_buffer_per_cpu *cpu_buffer)
1442 struct buffer_page *reader = cpu_buffer->reader_page;
1443 struct buffer_page *head = cpu_buffer->head_page;
1444 struct buffer_page *commit = cpu_buffer->commit_page;
1446 return reader->read == rb_page_commit(reader) &&
1447 (commit == reader ||
1449 head->read == rb_page_commit(commit)));
1453 * ring_buffer_record_disable - stop all writes into the buffer
1454 * @buffer: The ring buffer to stop writes to.
1456 * This prevents all writes to the buffer. Any attempt to write
1457 * to the buffer after this will fail and return NULL.
1459 * The caller should call synchronize_sched() after this.
1461 void ring_buffer_record_disable(struct ring_buffer *buffer)
1463 atomic_inc(&buffer->record_disabled);
1465 EXPORT_SYMBOL_GPL(ring_buffer_record_disable);
1468 * ring_buffer_record_enable - enable writes to the buffer
1469 * @buffer: The ring buffer to enable writes
1471 * Note, multiple disables will need the same number of enables
1472 * to truely enable the writing (much like preempt_disable).
1474 void ring_buffer_record_enable(struct ring_buffer *buffer)
1476 atomic_dec(&buffer->record_disabled);
1478 EXPORT_SYMBOL_GPL(ring_buffer_record_enable);
1481 * ring_buffer_record_disable_cpu - stop all writes into the cpu_buffer
1482 * @buffer: The ring buffer to stop writes to.
1483 * @cpu: The CPU buffer to stop
1485 * This prevents all writes to the buffer. Any attempt to write
1486 * to the buffer after this will fail and return NULL.
1488 * The caller should call synchronize_sched() after this.
1490 void ring_buffer_record_disable_cpu(struct ring_buffer *buffer, int cpu)
1492 struct ring_buffer_per_cpu *cpu_buffer;
1494 if (!cpumask_test_cpu(cpu, buffer->cpumask))
1497 cpu_buffer = buffer->buffers[cpu];
1498 atomic_inc(&cpu_buffer->record_disabled);
1500 EXPORT_SYMBOL_GPL(ring_buffer_record_disable_cpu);
1503 * ring_buffer_record_enable_cpu - enable writes to the buffer
1504 * @buffer: The ring buffer to enable writes
1505 * @cpu: The CPU to enable.
1507 * Note, multiple disables will need the same number of enables
1508 * to truely enable the writing (much like preempt_disable).
1510 void ring_buffer_record_enable_cpu(struct ring_buffer *buffer, int cpu)
1512 struct ring_buffer_per_cpu *cpu_buffer;
1514 if (!cpumask_test_cpu(cpu, buffer->cpumask))
1517 cpu_buffer = buffer->buffers[cpu];
1518 atomic_dec(&cpu_buffer->record_disabled);
1520 EXPORT_SYMBOL_GPL(ring_buffer_record_enable_cpu);
1523 * ring_buffer_entries_cpu - get the number of entries in a cpu buffer
1524 * @buffer: The ring buffer
1525 * @cpu: The per CPU buffer to get the entries from.
1527 unsigned long ring_buffer_entries_cpu(struct ring_buffer *buffer, int cpu)
1529 struct ring_buffer_per_cpu *cpu_buffer;
1531 if (!cpumask_test_cpu(cpu, buffer->cpumask))
1534 cpu_buffer = buffer->buffers[cpu];
1535 return cpu_buffer->entries;
1537 EXPORT_SYMBOL_GPL(ring_buffer_entries_cpu);
1540 * ring_buffer_overrun_cpu - get the number of overruns in a cpu_buffer
1541 * @buffer: The ring buffer
1542 * @cpu: The per CPU buffer to get the number of overruns from
1544 unsigned long ring_buffer_overrun_cpu(struct ring_buffer *buffer, int cpu)
1546 struct ring_buffer_per_cpu *cpu_buffer;
1548 if (!cpumask_test_cpu(cpu, buffer->cpumask))
1551 cpu_buffer = buffer->buffers[cpu];
1552 return cpu_buffer->overrun;
1554 EXPORT_SYMBOL_GPL(ring_buffer_overrun_cpu);
1557 * ring_buffer_entries - get the number of entries in a buffer
1558 * @buffer: The ring buffer
1560 * Returns the total number of entries in the ring buffer
1563 unsigned long ring_buffer_entries(struct ring_buffer *buffer)
1565 struct ring_buffer_per_cpu *cpu_buffer;
1566 unsigned long entries = 0;
1569 /* if you care about this being correct, lock the buffer */
1570 for_each_buffer_cpu(buffer, cpu) {
1571 cpu_buffer = buffer->buffers[cpu];
1572 entries += cpu_buffer->entries;
1577 EXPORT_SYMBOL_GPL(ring_buffer_entries);
1580 * ring_buffer_overrun_cpu - get the number of overruns in buffer
1581 * @buffer: The ring buffer
1583 * Returns the total number of overruns in the ring buffer
1586 unsigned long ring_buffer_overruns(struct ring_buffer *buffer)
1588 struct ring_buffer_per_cpu *cpu_buffer;
1589 unsigned long overruns = 0;
1592 /* if you care about this being correct, lock the buffer */
1593 for_each_buffer_cpu(buffer, cpu) {
1594 cpu_buffer = buffer->buffers[cpu];
1595 overruns += cpu_buffer->overrun;
1600 EXPORT_SYMBOL_GPL(ring_buffer_overruns);
1602 static void rb_iter_reset(struct ring_buffer_iter *iter)
1604 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
1606 /* Iterator usage is expected to have record disabled */
1607 if (list_empty(&cpu_buffer->reader_page->list)) {
1608 iter->head_page = cpu_buffer->head_page;
1609 iter->head = cpu_buffer->head_page->read;
1611 iter->head_page = cpu_buffer->reader_page;
1612 iter->head = cpu_buffer->reader_page->read;
1615 iter->read_stamp = cpu_buffer->read_stamp;
1617 iter->read_stamp = iter->head_page->page->time_stamp;
1621 * ring_buffer_iter_reset - reset an iterator
1622 * @iter: The iterator to reset
1624 * Resets the iterator, so that it will start from the beginning
1627 void ring_buffer_iter_reset(struct ring_buffer_iter *iter)
1629 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
1630 unsigned long flags;
1632 spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
1633 rb_iter_reset(iter);
1634 spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
1636 EXPORT_SYMBOL_GPL(ring_buffer_iter_reset);
1639 * ring_buffer_iter_empty - check if an iterator has no more to read
1640 * @iter: The iterator to check
1642 int ring_buffer_iter_empty(struct ring_buffer_iter *iter)
1644 struct ring_buffer_per_cpu *cpu_buffer;
1646 cpu_buffer = iter->cpu_buffer;
1648 return iter->head_page == cpu_buffer->commit_page &&
1649 iter->head == rb_commit_index(cpu_buffer);
1651 EXPORT_SYMBOL_GPL(ring_buffer_iter_empty);
1654 rb_update_read_stamp(struct ring_buffer_per_cpu *cpu_buffer,
1655 struct ring_buffer_event *event)
1659 switch (event->type) {
1660 case RINGBUF_TYPE_PADDING:
1663 case RINGBUF_TYPE_TIME_EXTEND:
1664 delta = event->array[0];
1666 delta += event->time_delta;
1667 cpu_buffer->read_stamp += delta;
1670 case RINGBUF_TYPE_TIME_STAMP:
1671 /* FIXME: not implemented */
1674 case RINGBUF_TYPE_DATA:
1675 cpu_buffer->read_stamp += event->time_delta;
1685 rb_update_iter_read_stamp(struct ring_buffer_iter *iter,
1686 struct ring_buffer_event *event)
1690 switch (event->type) {
1691 case RINGBUF_TYPE_PADDING:
1694 case RINGBUF_TYPE_TIME_EXTEND:
1695 delta = event->array[0];
1697 delta += event->time_delta;
1698 iter->read_stamp += delta;
1701 case RINGBUF_TYPE_TIME_STAMP:
1702 /* FIXME: not implemented */
1705 case RINGBUF_TYPE_DATA:
1706 iter->read_stamp += event->time_delta;
1715 static struct buffer_page *
1716 rb_get_reader_page(struct ring_buffer_per_cpu *cpu_buffer)
1718 struct buffer_page *reader = NULL;
1719 unsigned long flags;
1722 local_irq_save(flags);
1723 __raw_spin_lock(&cpu_buffer->lock);
1727 * This should normally only loop twice. But because the
1728 * start of the reader inserts an empty page, it causes
1729 * a case where we will loop three times. There should be no
1730 * reason to loop four times (that I know of).
1732 if (RB_WARN_ON(cpu_buffer, ++nr_loops > 3)) {
1737 reader = cpu_buffer->reader_page;
1739 /* If there's more to read, return this page */
1740 if (cpu_buffer->reader_page->read < rb_page_size(reader))
1743 /* Never should we have an index greater than the size */
1744 if (RB_WARN_ON(cpu_buffer,
1745 cpu_buffer->reader_page->read > rb_page_size(reader)))
1748 /* check if we caught up to the tail */
1750 if (cpu_buffer->commit_page == cpu_buffer->reader_page)
1754 * Splice the empty reader page into the list around the head.
1755 * Reset the reader page to size zero.
1758 reader = cpu_buffer->head_page;
1759 cpu_buffer->reader_page->list.next = reader->list.next;
1760 cpu_buffer->reader_page->list.prev = reader->list.prev;
1762 local_set(&cpu_buffer->reader_page->write, 0);
1763 local_set(&cpu_buffer->reader_page->page->commit, 0);
1765 /* Make the reader page now replace the head */
1766 reader->list.prev->next = &cpu_buffer->reader_page->list;
1767 reader->list.next->prev = &cpu_buffer->reader_page->list;
1770 * If the tail is on the reader, then we must set the head
1771 * to the inserted page, otherwise we set it one before.
1773 cpu_buffer->head_page = cpu_buffer->reader_page;
1775 if (cpu_buffer->commit_page != reader)
1776 rb_inc_page(cpu_buffer, &cpu_buffer->head_page);
1778 /* Finally update the reader page to the new head */
1779 cpu_buffer->reader_page = reader;
1780 rb_reset_reader_page(cpu_buffer);
1785 __raw_spin_unlock(&cpu_buffer->lock);
1786 local_irq_restore(flags);
1791 static void rb_advance_reader(struct ring_buffer_per_cpu *cpu_buffer)
1793 struct ring_buffer_event *event;
1794 struct buffer_page *reader;
1797 reader = rb_get_reader_page(cpu_buffer);
1799 /* This function should not be called when buffer is empty */
1800 if (RB_WARN_ON(cpu_buffer, !reader))
1803 event = rb_reader_event(cpu_buffer);
1805 if (event->type == RINGBUF_TYPE_DATA)
1806 cpu_buffer->entries--;
1808 rb_update_read_stamp(cpu_buffer, event);
1810 length = rb_event_length(event);
1811 cpu_buffer->reader_page->read += length;
1814 static void rb_advance_iter(struct ring_buffer_iter *iter)
1816 struct ring_buffer *buffer;
1817 struct ring_buffer_per_cpu *cpu_buffer;
1818 struct ring_buffer_event *event;
1821 cpu_buffer = iter->cpu_buffer;
1822 buffer = cpu_buffer->buffer;
1825 * Check if we are at the end of the buffer.
1827 if (iter->head >= rb_page_size(iter->head_page)) {
1828 if (RB_WARN_ON(buffer,
1829 iter->head_page == cpu_buffer->commit_page))
1835 event = rb_iter_head_event(iter);
1837 length = rb_event_length(event);
1840 * This should not be called to advance the header if we are
1841 * at the tail of the buffer.
1843 if (RB_WARN_ON(cpu_buffer,
1844 (iter->head_page == cpu_buffer->commit_page) &&
1845 (iter->head + length > rb_commit_index(cpu_buffer))))
1848 rb_update_iter_read_stamp(iter, event);
1850 iter->head += length;
1852 /* check for end of page padding */
1853 if ((iter->head >= rb_page_size(iter->head_page)) &&
1854 (iter->head_page != cpu_buffer->commit_page))
1855 rb_advance_iter(iter);
1858 static struct ring_buffer_event *
1859 rb_buffer_peek(struct ring_buffer *buffer, int cpu, u64 *ts)
1861 struct ring_buffer_per_cpu *cpu_buffer;
1862 struct ring_buffer_event *event;
1863 struct buffer_page *reader;
1866 if (!cpumask_test_cpu(cpu, buffer->cpumask))
1869 cpu_buffer = buffer->buffers[cpu];
1873 * We repeat when a timestamp is encountered. It is possible
1874 * to get multiple timestamps from an interrupt entering just
1875 * as one timestamp is about to be written. The max times
1876 * that this can happen is the number of nested interrupts we
1877 * can have. Nesting 10 deep of interrupts is clearly
1880 if (RB_WARN_ON(cpu_buffer, ++nr_loops > 10))
1883 reader = rb_get_reader_page(cpu_buffer);
1887 event = rb_reader_event(cpu_buffer);
1889 switch (event->type) {
1890 case RINGBUF_TYPE_PADDING:
1891 RB_WARN_ON(cpu_buffer, 1);
1892 rb_advance_reader(cpu_buffer);
1895 case RINGBUF_TYPE_TIME_EXTEND:
1896 /* Internal data, OK to advance */
1897 rb_advance_reader(cpu_buffer);
1900 case RINGBUF_TYPE_TIME_STAMP:
1901 /* FIXME: not implemented */
1902 rb_advance_reader(cpu_buffer);
1905 case RINGBUF_TYPE_DATA:
1907 *ts = cpu_buffer->read_stamp + event->time_delta;
1908 ring_buffer_normalize_time_stamp(cpu_buffer->cpu, ts);
1918 EXPORT_SYMBOL_GPL(ring_buffer_peek);
1920 static struct ring_buffer_event *
1921 rb_iter_peek(struct ring_buffer_iter *iter, u64 *ts)
1923 struct ring_buffer *buffer;
1924 struct ring_buffer_per_cpu *cpu_buffer;
1925 struct ring_buffer_event *event;
1928 if (ring_buffer_iter_empty(iter))
1931 cpu_buffer = iter->cpu_buffer;
1932 buffer = cpu_buffer->buffer;
1936 * We repeat when a timestamp is encountered. It is possible
1937 * to get multiple timestamps from an interrupt entering just
1938 * as one timestamp is about to be written. The max times
1939 * that this can happen is the number of nested interrupts we
1940 * can have. Nesting 10 deep of interrupts is clearly
1943 if (RB_WARN_ON(cpu_buffer, ++nr_loops > 10))
1946 if (rb_per_cpu_empty(cpu_buffer))
1949 event = rb_iter_head_event(iter);
1951 switch (event->type) {
1952 case RINGBUF_TYPE_PADDING:
1956 case RINGBUF_TYPE_TIME_EXTEND:
1957 /* Internal data, OK to advance */
1958 rb_advance_iter(iter);
1961 case RINGBUF_TYPE_TIME_STAMP:
1962 /* FIXME: not implemented */
1963 rb_advance_iter(iter);
1966 case RINGBUF_TYPE_DATA:
1968 *ts = iter->read_stamp + event->time_delta;
1969 ring_buffer_normalize_time_stamp(cpu_buffer->cpu, ts);
1979 EXPORT_SYMBOL_GPL(ring_buffer_iter_peek);
1982 * ring_buffer_peek - peek at the next event to be read
1983 * @buffer: The ring buffer to read
1984 * @cpu: The cpu to peak at
1985 * @ts: The timestamp counter of this event.
1987 * This will return the event that will be read next, but does
1988 * not consume the data.
1990 struct ring_buffer_event *
1991 ring_buffer_peek(struct ring_buffer *buffer, int cpu, u64 *ts)
1993 struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
1994 struct ring_buffer_event *event;
1995 unsigned long flags;
1997 spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
1998 event = rb_buffer_peek(buffer, cpu, ts);
1999 spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
2005 * ring_buffer_iter_peek - peek at the next event to be read
2006 * @iter: The ring buffer iterator
2007 * @ts: The timestamp counter of this event.
2009 * This will return the event that will be read next, but does
2010 * not increment the iterator.
2012 struct ring_buffer_event *
2013 ring_buffer_iter_peek(struct ring_buffer_iter *iter, u64 *ts)
2015 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
2016 struct ring_buffer_event *event;
2017 unsigned long flags;
2019 spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
2020 event = rb_iter_peek(iter, ts);
2021 spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
2027 * ring_buffer_consume - return an event and consume it
2028 * @buffer: The ring buffer to get the next event from
2030 * Returns the next event in the ring buffer, and that event is consumed.
2031 * Meaning, that sequential reads will keep returning a different event,
2032 * and eventually empty the ring buffer if the producer is slower.
2034 struct ring_buffer_event *
2035 ring_buffer_consume(struct ring_buffer *buffer, int cpu, u64 *ts)
2037 struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
2038 struct ring_buffer_event *event;
2039 unsigned long flags;
2041 if (!cpumask_test_cpu(cpu, buffer->cpumask))
2044 spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
2046 event = rb_buffer_peek(buffer, cpu, ts);
2050 rb_advance_reader(cpu_buffer);
2053 spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
2057 EXPORT_SYMBOL_GPL(ring_buffer_consume);
2060 * ring_buffer_read_start - start a non consuming read of the buffer
2061 * @buffer: The ring buffer to read from
2062 * @cpu: The cpu buffer to iterate over
2064 * This starts up an iteration through the buffer. It also disables
2065 * the recording to the buffer until the reading is finished.
2066 * This prevents the reading from being corrupted. This is not
2067 * a consuming read, so a producer is not expected.
2069 * Must be paired with ring_buffer_finish.
2071 struct ring_buffer_iter *
2072 ring_buffer_read_start(struct ring_buffer *buffer, int cpu)
2074 struct ring_buffer_per_cpu *cpu_buffer;
2075 struct ring_buffer_iter *iter;
2076 unsigned long flags;
2078 if (!cpumask_test_cpu(cpu, buffer->cpumask))
2081 iter = kmalloc(sizeof(*iter), GFP_KERNEL);
2085 cpu_buffer = buffer->buffers[cpu];
2087 iter->cpu_buffer = cpu_buffer;
2089 atomic_inc(&cpu_buffer->record_disabled);
2090 synchronize_sched();
2092 spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
2093 __raw_spin_lock(&cpu_buffer->lock);
2094 rb_iter_reset(iter);
2095 __raw_spin_unlock(&cpu_buffer->lock);
2096 spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
2100 EXPORT_SYMBOL_GPL(ring_buffer_read_start);
2103 * ring_buffer_finish - finish reading the iterator of the buffer
2104 * @iter: The iterator retrieved by ring_buffer_start
2106 * This re-enables the recording to the buffer, and frees the
2110 ring_buffer_read_finish(struct ring_buffer_iter *iter)
2112 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
2114 atomic_dec(&cpu_buffer->record_disabled);
2117 EXPORT_SYMBOL_GPL(ring_buffer_read_finish);
2120 * ring_buffer_read - read the next item in the ring buffer by the iterator
2121 * @iter: The ring buffer iterator
2122 * @ts: The time stamp of the event read.
2124 * This reads the next event in the ring buffer and increments the iterator.
2126 struct ring_buffer_event *
2127 ring_buffer_read(struct ring_buffer_iter *iter, u64 *ts)
2129 struct ring_buffer_event *event;
2130 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
2131 unsigned long flags;
2133 spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
2134 event = rb_iter_peek(iter, ts);
2138 rb_advance_iter(iter);
2140 spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
2144 EXPORT_SYMBOL_GPL(ring_buffer_read);
2147 * ring_buffer_size - return the size of the ring buffer (in bytes)
2148 * @buffer: The ring buffer.
2150 unsigned long ring_buffer_size(struct ring_buffer *buffer)
2152 return BUF_PAGE_SIZE * buffer->pages;
2154 EXPORT_SYMBOL_GPL(ring_buffer_size);
2157 rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer)
2159 cpu_buffer->head_page
2160 = list_entry(cpu_buffer->pages.next, struct buffer_page, list);
2161 local_set(&cpu_buffer->head_page->write, 0);
2162 local_set(&cpu_buffer->head_page->page->commit, 0);
2164 cpu_buffer->head_page->read = 0;
2166 cpu_buffer->tail_page = cpu_buffer->head_page;
2167 cpu_buffer->commit_page = cpu_buffer->head_page;
2169 INIT_LIST_HEAD(&cpu_buffer->reader_page->list);
2170 local_set(&cpu_buffer->reader_page->write, 0);
2171 local_set(&cpu_buffer->reader_page->page->commit, 0);
2172 cpu_buffer->reader_page->read = 0;
2174 cpu_buffer->overrun = 0;
2175 cpu_buffer->entries = 0;
2179 * ring_buffer_reset_cpu - reset a ring buffer per CPU buffer
2180 * @buffer: The ring buffer to reset a per cpu buffer of
2181 * @cpu: The CPU buffer to be reset
2183 void ring_buffer_reset_cpu(struct ring_buffer *buffer, int cpu)
2185 struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
2186 unsigned long flags;
2188 if (!cpumask_test_cpu(cpu, buffer->cpumask))
2191 spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
2193 __raw_spin_lock(&cpu_buffer->lock);
2195 rb_reset_cpu(cpu_buffer);
2197 __raw_spin_unlock(&cpu_buffer->lock);
2199 spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
2201 EXPORT_SYMBOL_GPL(ring_buffer_reset_cpu);
2204 * ring_buffer_reset - reset a ring buffer
2205 * @buffer: The ring buffer to reset all cpu buffers
2207 void ring_buffer_reset(struct ring_buffer *buffer)
2211 for_each_buffer_cpu(buffer, cpu)
2212 ring_buffer_reset_cpu(buffer, cpu);
2214 EXPORT_SYMBOL_GPL(ring_buffer_reset);
2217 * rind_buffer_empty - is the ring buffer empty?
2218 * @buffer: The ring buffer to test
2220 int ring_buffer_empty(struct ring_buffer *buffer)
2222 struct ring_buffer_per_cpu *cpu_buffer;
2225 /* yes this is racy, but if you don't like the race, lock the buffer */
2226 for_each_buffer_cpu(buffer, cpu) {
2227 cpu_buffer = buffer->buffers[cpu];
2228 if (!rb_per_cpu_empty(cpu_buffer))
2233 EXPORT_SYMBOL_GPL(ring_buffer_empty);
2236 * ring_buffer_empty_cpu - is a cpu buffer of a ring buffer empty?
2237 * @buffer: The ring buffer
2238 * @cpu: The CPU buffer to test
2240 int ring_buffer_empty_cpu(struct ring_buffer *buffer, int cpu)
2242 struct ring_buffer_per_cpu *cpu_buffer;
2244 if (!cpumask_test_cpu(cpu, buffer->cpumask))
2247 cpu_buffer = buffer->buffers[cpu];
2248 return rb_per_cpu_empty(cpu_buffer);
2250 EXPORT_SYMBOL_GPL(ring_buffer_empty_cpu);
2253 * ring_buffer_swap_cpu - swap a CPU buffer between two ring buffers
2254 * @buffer_a: One buffer to swap with
2255 * @buffer_b: The other buffer to swap with
2257 * This function is useful for tracers that want to take a "snapshot"
2258 * of a CPU buffer and has another back up buffer lying around.
2259 * it is expected that the tracer handles the cpu buffer not being
2260 * used at the moment.
2262 int ring_buffer_swap_cpu(struct ring_buffer *buffer_a,
2263 struct ring_buffer *buffer_b, int cpu)
2265 struct ring_buffer_per_cpu *cpu_buffer_a;
2266 struct ring_buffer_per_cpu *cpu_buffer_b;
2268 if (!cpumask_test_cpu(cpu, buffer_a->cpumask) ||
2269 !cpumask_test_cpu(cpu, buffer_b->cpumask))
2272 /* At least make sure the two buffers are somewhat the same */
2273 if (buffer_a->pages != buffer_b->pages)
2276 cpu_buffer_a = buffer_a->buffers[cpu];
2277 cpu_buffer_b = buffer_b->buffers[cpu];
2280 * We can't do a synchronize_sched here because this
2281 * function can be called in atomic context.
2282 * Normally this will be called from the same CPU as cpu.
2283 * If not it's up to the caller to protect this.
2285 atomic_inc(&cpu_buffer_a->record_disabled);
2286 atomic_inc(&cpu_buffer_b->record_disabled);
2288 buffer_a->buffers[cpu] = cpu_buffer_b;
2289 buffer_b->buffers[cpu] = cpu_buffer_a;
2291 cpu_buffer_b->buffer = buffer_a;
2292 cpu_buffer_a->buffer = buffer_b;
2294 atomic_dec(&cpu_buffer_a->record_disabled);
2295 atomic_dec(&cpu_buffer_b->record_disabled);
2299 EXPORT_SYMBOL_GPL(ring_buffer_swap_cpu);
2301 static void rb_remove_entries(struct ring_buffer_per_cpu *cpu_buffer,
2302 struct buffer_data_page *bpage)
2304 struct ring_buffer_event *event;
2307 __raw_spin_lock(&cpu_buffer->lock);
2308 for (head = 0; head < local_read(&bpage->commit);
2309 head += rb_event_length(event)) {
2311 event = __rb_data_page_index(bpage, head);
2312 if (RB_WARN_ON(cpu_buffer, rb_null_event(event)))
2314 /* Only count data entries */
2315 if (event->type != RINGBUF_TYPE_DATA)
2317 cpu_buffer->entries--;
2319 __raw_spin_unlock(&cpu_buffer->lock);
2323 * ring_buffer_alloc_read_page - allocate a page to read from buffer
2324 * @buffer: the buffer to allocate for.
2326 * This function is used in conjunction with ring_buffer_read_page.
2327 * When reading a full page from the ring buffer, these functions
2328 * can be used to speed up the process. The calling function should
2329 * allocate a few pages first with this function. Then when it
2330 * needs to get pages from the ring buffer, it passes the result
2331 * of this function into ring_buffer_read_page, which will swap
2332 * the page that was allocated, with the read page of the buffer.
2335 * The page allocated, or NULL on error.
2337 void *ring_buffer_alloc_read_page(struct ring_buffer *buffer)
2340 struct buffer_data_page *bpage;
2342 addr = __get_free_page(GFP_KERNEL);
2346 bpage = (void *)addr;
2352 * ring_buffer_free_read_page - free an allocated read page
2353 * @buffer: the buffer the page was allocate for
2354 * @data: the page to free
2356 * Free a page allocated from ring_buffer_alloc_read_page.
2358 void ring_buffer_free_read_page(struct ring_buffer *buffer, void *data)
2360 free_page((unsigned long)data);
2364 * ring_buffer_read_page - extract a page from the ring buffer
2365 * @buffer: buffer to extract from
2366 * @data_page: the page to use allocated from ring_buffer_alloc_read_page
2367 * @cpu: the cpu of the buffer to extract
2368 * @full: should the extraction only happen when the page is full.
2370 * This function will pull out a page from the ring buffer and consume it.
2371 * @data_page must be the address of the variable that was returned
2372 * from ring_buffer_alloc_read_page. This is because the page might be used
2373 * to swap with a page in the ring buffer.
2376 * rpage = ring_buffer_alloc_page(buffer);
2379 * ret = ring_buffer_read_page(buffer, &rpage, cpu, 0);
2381 * process_page(rpage);
2383 * When @full is set, the function will not return true unless
2384 * the writer is off the reader page.
2386 * Note: it is up to the calling functions to handle sleeps and wakeups.
2387 * The ring buffer can be used anywhere in the kernel and can not
2388 * blindly call wake_up. The layer that uses the ring buffer must be
2389 * responsible for that.
2392 * 1 if data has been transferred
2393 * 0 if no data has been transferred.
2395 int ring_buffer_read_page(struct ring_buffer *buffer,
2396 void **data_page, int cpu, int full)
2398 struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
2399 struct ring_buffer_event *event;
2400 struct buffer_data_page *bpage;
2401 unsigned long flags;
2411 spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
2414 * rb_buffer_peek will get the next ring buffer if
2415 * the current reader page is empty.
2417 event = rb_buffer_peek(buffer, cpu, NULL);
2421 /* check for data */
2422 if (!local_read(&cpu_buffer->reader_page->page->commit))
2425 * If the writer is already off of the read page, then simply
2426 * switch the read page with the given page. Otherwise
2427 * we need to copy the data from the reader to the writer.
2429 if (cpu_buffer->reader_page == cpu_buffer->commit_page) {
2430 unsigned int read = cpu_buffer->reader_page->read;
2434 /* The writer is still on the reader page, we must copy */
2435 bpage = cpu_buffer->reader_page->page;
2437 cpu_buffer->reader_page->page->data + read,
2438 local_read(&bpage->commit) - read);
2440 /* consume what was read */
2441 cpu_buffer->reader_page += read;
2444 /* swap the pages */
2445 rb_init_page(bpage);
2446 bpage = cpu_buffer->reader_page->page;
2447 cpu_buffer->reader_page->page = *data_page;
2448 cpu_buffer->reader_page->read = 0;
2453 /* update the entry counter */
2454 rb_remove_entries(cpu_buffer, bpage);
2456 spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
2462 rb_simple_read(struct file *filp, char __user *ubuf,
2463 size_t cnt, loff_t *ppos)
2465 long *p = filp->private_data;
2469 if (test_bit(RB_BUFFERS_DISABLED_BIT, p))
2470 r = sprintf(buf, "permanently disabled\n");
2472 r = sprintf(buf, "%d\n", test_bit(RB_BUFFERS_ON_BIT, p));
2474 return simple_read_from_buffer(ubuf, cnt, ppos, buf, r);
2478 rb_simple_write(struct file *filp, const char __user *ubuf,
2479 size_t cnt, loff_t *ppos)
2481 long *p = filp->private_data;
2486 if (cnt >= sizeof(buf))
2489 if (copy_from_user(&buf, ubuf, cnt))
2494 ret = strict_strtoul(buf, 10, &val);
2499 set_bit(RB_BUFFERS_ON_BIT, p);
2501 clear_bit(RB_BUFFERS_ON_BIT, p);
2508 static struct file_operations rb_simple_fops = {
2509 .open = tracing_open_generic,
2510 .read = rb_simple_read,
2511 .write = rb_simple_write,
2515 static __init int rb_init_debugfs(void)
2517 struct dentry *d_tracer;
2518 struct dentry *entry;
2520 d_tracer = tracing_init_dentry();
2522 entry = debugfs_create_file("tracing_on", 0644, d_tracer,
2523 &ring_buffer_flags, &rb_simple_fops);
2525 pr_warning("Could not create debugfs 'tracing_on' entry\n");
2530 fs_initcall(rb_init_debugfs);