4 * Copyright (C) 2008 Steven Rostedt <srostedt@redhat.com>
6 #include <linux/ring_buffer.h>
7 #include <linux/trace_clock.h>
8 #include <linux/ftrace_irq.h>
9 #include <linux/spinlock.h>
10 #include <linux/debugfs.h>
11 #include <linux/uaccess.h>
12 #include <linux/hardirq.h>
13 #include <linux/kmemcheck.h>
14 #include <linux/module.h>
15 #include <linux/percpu.h>
16 #include <linux/mutex.h>
17 #include <linux/init.h>
18 #include <linux/hash.h>
19 #include <linux/list.h>
20 #include <linux/cpu.h>
26 * The ring buffer header is special. We must manually up keep it.
28 int ring_buffer_print_entry_header(struct trace_seq
*s
)
32 ret
= trace_seq_printf(s
, "# compressed entry header\n");
33 ret
= trace_seq_printf(s
, "\ttype_len : 5 bits\n");
34 ret
= trace_seq_printf(s
, "\ttime_delta : 27 bits\n");
35 ret
= trace_seq_printf(s
, "\tarray : 32 bits\n");
36 ret
= trace_seq_printf(s
, "\n");
37 ret
= trace_seq_printf(s
, "\tpadding : type == %d\n",
38 RINGBUF_TYPE_PADDING
);
39 ret
= trace_seq_printf(s
, "\ttime_extend : type == %d\n",
40 RINGBUF_TYPE_TIME_EXTEND
);
41 ret
= trace_seq_printf(s
, "\tdata max type_len == %d\n",
42 RINGBUF_TYPE_DATA_TYPE_LEN_MAX
);
48 * The ring buffer is made up of a list of pages. A separate list of pages is
49 * allocated for each CPU. A writer may only write to a buffer that is
50 * associated with the CPU it is currently executing on. A reader may read
51 * from any per cpu buffer.
53 * The reader is special. For each per cpu buffer, the reader has its own
54 * reader page. When a reader has read the entire reader page, this reader
55 * page is swapped with another page in the ring buffer.
57 * Now, as long as the writer is off the reader page, the reader can do what
58 * ever it wants with that page. The writer will never write to that page
59 * again (as long as it is out of the ring buffer).
61 * Here's some silly ASCII art.
64 * |reader| RING BUFFER
66 * +------+ +---+ +---+ +---+
75 * |reader| RING BUFFER
76 * |page |------------------v
77 * +------+ +---+ +---+ +---+
86 * |reader| RING BUFFER
87 * |page |------------------v
88 * +------+ +---+ +---+ +---+
93 * +------------------------------+
97 * |buffer| RING BUFFER
98 * |page |------------------v
99 * +------+ +---+ +---+ +---+
101 * | New +---+ +---+ +---+
104 * +------------------------------+
107 * After we make this swap, the reader can hand this page off to the splice
108 * code and be done with it. It can even allocate a new page if it needs to
109 * and swap that into the ring buffer.
111 * We will be using cmpxchg soon to make all this lockless.
116 * A fast way to enable or disable all ring buffers is to
117 * call tracing_on or tracing_off. Turning off the ring buffers
118 * prevents all ring buffers from being recorded to.
119 * Turning this switch on, makes it OK to write to the
120 * ring buffer, if the ring buffer is enabled itself.
122 * There's three layers that must be on in order to write
123 * to the ring buffer.
125 * 1) This global flag must be set.
126 * 2) The ring buffer must be enabled for recording.
127 * 3) The per cpu buffer must be enabled for recording.
129 * In case of an anomaly, this global flag has a bit set that
130 * will permantly disable all ring buffers.
134 * Global flag to disable all recording to ring buffers
135 * This has two bits: ON, DISABLED
139 * 0 0 : ring buffers are off
140 * 1 0 : ring buffers are on
141 * X 1 : ring buffers are permanently disabled
145 RB_BUFFERS_ON_BIT
= 0,
146 RB_BUFFERS_DISABLED_BIT
= 1,
150 RB_BUFFERS_ON
= 1 << RB_BUFFERS_ON_BIT
,
151 RB_BUFFERS_DISABLED
= 1 << RB_BUFFERS_DISABLED_BIT
,
154 static unsigned long ring_buffer_flags __read_mostly
= RB_BUFFERS_ON
;
156 #define BUF_PAGE_HDR_SIZE offsetof(struct buffer_data_page, data)
159 * tracing_on - enable all tracing buffers
161 * This function enables all tracing buffers that may have been
162 * disabled with tracing_off.
164 void tracing_on(void)
166 set_bit(RB_BUFFERS_ON_BIT
, &ring_buffer_flags
);
168 EXPORT_SYMBOL_GPL(tracing_on
);
171 * tracing_off - turn off all tracing buffers
173 * This function stops all tracing buffers from recording data.
174 * It does not disable any overhead the tracers themselves may
175 * be causing. This function simply causes all recording to
176 * the ring buffers to fail.
178 void tracing_off(void)
180 clear_bit(RB_BUFFERS_ON_BIT
, &ring_buffer_flags
);
182 EXPORT_SYMBOL_GPL(tracing_off
);
185 * tracing_off_permanent - permanently disable ring buffers
187 * This function, once called, will disable all ring buffers
190 void tracing_off_permanent(void)
192 set_bit(RB_BUFFERS_DISABLED_BIT
, &ring_buffer_flags
);
196 * tracing_is_on - show state of ring buffers enabled
198 int tracing_is_on(void)
200 return ring_buffer_flags
== RB_BUFFERS_ON
;
202 EXPORT_SYMBOL_GPL(tracing_is_on
);
204 #define RB_EVNT_HDR_SIZE (offsetof(struct ring_buffer_event, array))
205 #define RB_ALIGNMENT 4U
206 #define RB_MAX_SMALL_DATA (RB_ALIGNMENT * RINGBUF_TYPE_DATA_TYPE_LEN_MAX)
207 #define RB_EVNT_MIN_SIZE 8U /* two 32bit words */
209 /* define RINGBUF_TYPE_DATA for 'case RINGBUF_TYPE_DATA:' */
210 #define RINGBUF_TYPE_DATA 0 ... RINGBUF_TYPE_DATA_TYPE_LEN_MAX
213 RB_LEN_TIME_EXTEND
= 8,
214 RB_LEN_TIME_STAMP
= 16,
217 static inline int rb_null_event(struct ring_buffer_event
*event
)
219 return event
->type_len
== RINGBUF_TYPE_PADDING
&& !event
->time_delta
;
222 static void rb_event_set_padding(struct ring_buffer_event
*event
)
224 /* padding has a NULL time_delta */
225 event
->type_len
= RINGBUF_TYPE_PADDING
;
226 event
->time_delta
= 0;
230 rb_event_data_length(struct ring_buffer_event
*event
)
235 length
= event
->type_len
* RB_ALIGNMENT
;
237 length
= event
->array
[0];
238 return length
+ RB_EVNT_HDR_SIZE
;
241 /* inline for ring buffer fast paths */
243 rb_event_length(struct ring_buffer_event
*event
)
245 switch (event
->type_len
) {
246 case RINGBUF_TYPE_PADDING
:
247 if (rb_null_event(event
))
250 return event
->array
[0] + RB_EVNT_HDR_SIZE
;
252 case RINGBUF_TYPE_TIME_EXTEND
:
253 return RB_LEN_TIME_EXTEND
;
255 case RINGBUF_TYPE_TIME_STAMP
:
256 return RB_LEN_TIME_STAMP
;
258 case RINGBUF_TYPE_DATA
:
259 return rb_event_data_length(event
);
268 * ring_buffer_event_length - return the length of the event
269 * @event: the event to get the length of
271 unsigned ring_buffer_event_length(struct ring_buffer_event
*event
)
273 unsigned length
= rb_event_length(event
);
274 if (event
->type_len
> RINGBUF_TYPE_DATA_TYPE_LEN_MAX
)
276 length
-= RB_EVNT_HDR_SIZE
;
277 if (length
> RB_MAX_SMALL_DATA
+ sizeof(event
->array
[0]))
278 length
-= sizeof(event
->array
[0]);
281 EXPORT_SYMBOL_GPL(ring_buffer_event_length
);
283 /* inline for ring buffer fast paths */
285 rb_event_data(struct ring_buffer_event
*event
)
287 BUG_ON(event
->type_len
> RINGBUF_TYPE_DATA_TYPE_LEN_MAX
);
288 /* If length is in len field, then array[0] has the data */
290 return (void *)&event
->array
[0];
291 /* Otherwise length is in array[0] and array[1] has the data */
292 return (void *)&event
->array
[1];
296 * ring_buffer_event_data - return the data of the event
297 * @event: the event to get the data from
299 void *ring_buffer_event_data(struct ring_buffer_event
*event
)
301 return rb_event_data(event
);
303 EXPORT_SYMBOL_GPL(ring_buffer_event_data
);
305 #define for_each_buffer_cpu(buffer, cpu) \
306 for_each_cpu(cpu, buffer->cpumask)
309 #define TS_MASK ((1ULL << TS_SHIFT) - 1)
310 #define TS_DELTA_TEST (~TS_MASK)
312 struct buffer_data_page
{
313 u64 time_stamp
; /* page time stamp */
314 local_t commit
; /* write committed index */
315 unsigned char data
[]; /* data of buffer page */
319 * Note, the buffer_page list must be first. The buffer pages
320 * are allocated in cache lines, which means that each buffer
321 * page will be at the beginning of a cache line, and thus
322 * the least significant bits will be zero. We use this to
323 * add flags in the list struct pointers, to make the ring buffer
327 struct list_head list
; /* list of buffer pages */
328 local_t write
; /* index for next write */
329 unsigned read
; /* index for next read */
330 local_t entries
; /* entries on this page */
331 struct buffer_data_page
*page
; /* Actual data page */
335 * The buffer page counters, write and entries, must be reset
336 * atomically when crossing page boundaries. To synchronize this
337 * update, two counters are inserted into the number. One is
338 * the actual counter for the write position or count on the page.
340 * The other is a counter of updaters. Before an update happens
341 * the update partition of the counter is incremented. This will
342 * allow the updater to update the counter atomically.
344 * The counter is 20 bits, and the state data is 12.
346 #define RB_WRITE_MASK 0xfffff
347 #define RB_WRITE_INTCNT (1 << 20)
349 static void rb_init_page(struct buffer_data_page
*bpage
)
351 local_set(&bpage
->commit
, 0);
355 * ring_buffer_page_len - the size of data on the page.
356 * @page: The page to read
358 * Returns the amount of data on the page, including buffer page header.
360 size_t ring_buffer_page_len(void *page
)
362 return local_read(&((struct buffer_data_page
*)page
)->commit
)
367 * Also stolen from mm/slob.c. Thanks to Mathieu Desnoyers for pointing
370 static void free_buffer_page(struct buffer_page
*bpage
)
372 free_page((unsigned long)bpage
->page
);
377 * We need to fit the time_stamp delta into 27 bits.
379 static inline int test_time_stamp(u64 delta
)
381 if (delta
& TS_DELTA_TEST
)
386 #define BUF_PAGE_SIZE (PAGE_SIZE - BUF_PAGE_HDR_SIZE)
388 /* Max payload is BUF_PAGE_SIZE - header (8bytes) */
389 #define BUF_MAX_DATA_SIZE (BUF_PAGE_SIZE - (sizeof(u32) * 2))
391 /* Max number of timestamps that can fit on a page */
392 #define RB_TIMESTAMPS_PER_PAGE (BUF_PAGE_SIZE / RB_LEN_TIME_STAMP)
394 int ring_buffer_print_page_header(struct trace_seq
*s
)
396 struct buffer_data_page field
;
399 ret
= trace_seq_printf(s
, "\tfield: u64 timestamp;\t"
400 "offset:0;\tsize:%u;\tsigned:%u;\n",
401 (unsigned int)sizeof(field
.time_stamp
),
402 (unsigned int)is_signed_type(u64
));
404 ret
= trace_seq_printf(s
, "\tfield: local_t commit;\t"
405 "offset:%u;\tsize:%u;\tsigned:%u;\n",
406 (unsigned int)offsetof(typeof(field
), commit
),
407 (unsigned int)sizeof(field
.commit
),
408 (unsigned int)is_signed_type(long));
410 ret
= trace_seq_printf(s
, "\tfield: char data;\t"
411 "offset:%u;\tsize:%u;\tsigned:%u;\n",
412 (unsigned int)offsetof(typeof(field
), data
),
413 (unsigned int)BUF_PAGE_SIZE
,
414 (unsigned int)is_signed_type(char));
420 * head_page == tail_page && head == tail then buffer is empty.
422 struct ring_buffer_per_cpu
{
424 struct ring_buffer
*buffer
;
425 spinlock_t reader_lock
; /* serialize readers */
427 struct lock_class_key lock_key
;
428 struct list_head
*pages
;
429 struct buffer_page
*head_page
; /* read from head */
430 struct buffer_page
*tail_page
; /* write to tail */
431 struct buffer_page
*commit_page
; /* committed pages */
432 struct buffer_page
*reader_page
;
433 local_t commit_overrun
;
441 atomic_t record_disabled
;
448 atomic_t record_disabled
;
449 cpumask_var_t cpumask
;
451 struct lock_class_key
*reader_lock_key
;
455 struct ring_buffer_per_cpu
**buffers
;
457 #ifdef CONFIG_HOTPLUG_CPU
458 struct notifier_block cpu_notify
;
463 struct ring_buffer_iter
{
464 struct ring_buffer_per_cpu
*cpu_buffer
;
466 struct buffer_page
*head_page
;
470 /* buffer may be either ring_buffer or ring_buffer_per_cpu */
471 #define RB_WARN_ON(b, cond) \
473 int _____ret = unlikely(cond); \
475 if (__same_type(*(b), struct ring_buffer_per_cpu)) { \
476 struct ring_buffer_per_cpu *__b = \
478 atomic_inc(&__b->buffer->record_disabled); \
480 atomic_inc(&b->record_disabled); \
486 /* Up this if you want to test the TIME_EXTENTS and normalization */
487 #define DEBUG_SHIFT 0
489 static inline u64
rb_time_stamp(struct ring_buffer
*buffer
, int cpu
)
491 /* shift to debug/test normalization and TIME_EXTENTS */
492 return buffer
->clock() << DEBUG_SHIFT
;
495 u64
ring_buffer_time_stamp(struct ring_buffer
*buffer
, int cpu
)
499 preempt_disable_notrace();
500 time
= rb_time_stamp(buffer
, cpu
);
501 preempt_enable_no_resched_notrace();
505 EXPORT_SYMBOL_GPL(ring_buffer_time_stamp
);
507 void ring_buffer_normalize_time_stamp(struct ring_buffer
*buffer
,
510 /* Just stupid testing the normalize function and deltas */
513 EXPORT_SYMBOL_GPL(ring_buffer_normalize_time_stamp
);
516 * Making the ring buffer lockless makes things tricky.
517 * Although writes only happen on the CPU that they are on,
518 * and they only need to worry about interrupts. Reads can
521 * The reader page is always off the ring buffer, but when the
522 * reader finishes with a page, it needs to swap its page with
523 * a new one from the buffer. The reader needs to take from
524 * the head (writes go to the tail). But if a writer is in overwrite
525 * mode and wraps, it must push the head page forward.
527 * Here lies the problem.
529 * The reader must be careful to replace only the head page, and
530 * not another one. As described at the top of the file in the
531 * ASCII art, the reader sets its old page to point to the next
532 * page after head. It then sets the page after head to point to
533 * the old reader page. But if the writer moves the head page
534 * during this operation, the reader could end up with the tail.
536 * We use cmpxchg to help prevent this race. We also do something
537 * special with the page before head. We set the LSB to 1.
539 * When the writer must push the page forward, it will clear the
540 * bit that points to the head page, move the head, and then set
541 * the bit that points to the new head page.
543 * We also don't want an interrupt coming in and moving the head
544 * page on another writer. Thus we use the second LSB to catch
547 * head->list->prev->next bit 1 bit 0
550 * Points to head page 0 1
553 * Note we can not trust the prev pointer of the head page, because:
555 * +----+ +-----+ +-----+
556 * | |------>| T |---X--->| N |
558 * +----+ +-----+ +-----+
561 * +----------| R |----------+ |
565 * Key: ---X--> HEAD flag set in pointer
570 * (see __rb_reserve_next() to see where this happens)
572 * What the above shows is that the reader just swapped out
573 * the reader page with a page in the buffer, but before it
574 * could make the new header point back to the new page added
575 * it was preempted by a writer. The writer moved forward onto
576 * the new page added by the reader and is about to move forward
579 * You can see, it is legitimate for the previous pointer of
580 * the head (or any page) not to point back to itself. But only
584 #define RB_PAGE_NORMAL 0UL
585 #define RB_PAGE_HEAD 1UL
586 #define RB_PAGE_UPDATE 2UL
589 #define RB_FLAG_MASK 3UL
591 /* PAGE_MOVED is not part of the mask */
592 #define RB_PAGE_MOVED 4UL
595 * rb_list_head - remove any bit
597 static struct list_head
*rb_list_head(struct list_head
*list
)
599 unsigned long val
= (unsigned long)list
;
601 return (struct list_head
*)(val
& ~RB_FLAG_MASK
);
605 * rb_is_head_page - test if the give page is the head page
607 * Because the reader may move the head_page pointer, we can
608 * not trust what the head page is (it may be pointing to
609 * the reader page). But if the next page is a header page,
610 * its flags will be non zero.
613 rb_is_head_page(struct ring_buffer_per_cpu
*cpu_buffer
,
614 struct buffer_page
*page
, struct list_head
*list
)
618 val
= (unsigned long)list
->next
;
620 if ((val
& ~RB_FLAG_MASK
) != (unsigned long)&page
->list
)
621 return RB_PAGE_MOVED
;
623 return val
& RB_FLAG_MASK
;
629 * The unique thing about the reader page, is that, if the
630 * writer is ever on it, the previous pointer never points
631 * back to the reader page.
633 static int rb_is_reader_page(struct buffer_page
*page
)
635 struct list_head
*list
= page
->list
.prev
;
637 return rb_list_head(list
->next
) != &page
->list
;
641 * rb_set_list_to_head - set a list_head to be pointing to head.
643 static void rb_set_list_to_head(struct ring_buffer_per_cpu
*cpu_buffer
,
644 struct list_head
*list
)
648 ptr
= (unsigned long *)&list
->next
;
649 *ptr
|= RB_PAGE_HEAD
;
650 *ptr
&= ~RB_PAGE_UPDATE
;
654 * rb_head_page_activate - sets up head page
656 static void rb_head_page_activate(struct ring_buffer_per_cpu
*cpu_buffer
)
658 struct buffer_page
*head
;
660 head
= cpu_buffer
->head_page
;
665 * Set the previous list pointer to have the HEAD flag.
667 rb_set_list_to_head(cpu_buffer
, head
->list
.prev
);
670 static void rb_list_head_clear(struct list_head
*list
)
672 unsigned long *ptr
= (unsigned long *)&list
->next
;
674 *ptr
&= ~RB_FLAG_MASK
;
678 * rb_head_page_dactivate - clears head page ptr (for free list)
681 rb_head_page_deactivate(struct ring_buffer_per_cpu
*cpu_buffer
)
683 struct list_head
*hd
;
685 /* Go through the whole list and clear any pointers found. */
686 rb_list_head_clear(cpu_buffer
->pages
);
688 list_for_each(hd
, cpu_buffer
->pages
)
689 rb_list_head_clear(hd
);
692 static int rb_head_page_set(struct ring_buffer_per_cpu
*cpu_buffer
,
693 struct buffer_page
*head
,
694 struct buffer_page
*prev
,
695 int old_flag
, int new_flag
)
697 struct list_head
*list
;
698 unsigned long val
= (unsigned long)&head
->list
;
703 val
&= ~RB_FLAG_MASK
;
705 ret
= cmpxchg((unsigned long *)&list
->next
,
706 val
| old_flag
, val
| new_flag
);
708 /* check if the reader took the page */
709 if ((ret
& ~RB_FLAG_MASK
) != val
)
710 return RB_PAGE_MOVED
;
712 return ret
& RB_FLAG_MASK
;
715 static int rb_head_page_set_update(struct ring_buffer_per_cpu
*cpu_buffer
,
716 struct buffer_page
*head
,
717 struct buffer_page
*prev
,
720 return rb_head_page_set(cpu_buffer
, head
, prev
,
721 old_flag
, RB_PAGE_UPDATE
);
724 static int rb_head_page_set_head(struct ring_buffer_per_cpu
*cpu_buffer
,
725 struct buffer_page
*head
,
726 struct buffer_page
*prev
,
729 return rb_head_page_set(cpu_buffer
, head
, prev
,
730 old_flag
, RB_PAGE_HEAD
);
733 static int rb_head_page_set_normal(struct ring_buffer_per_cpu
*cpu_buffer
,
734 struct buffer_page
*head
,
735 struct buffer_page
*prev
,
738 return rb_head_page_set(cpu_buffer
, head
, prev
,
739 old_flag
, RB_PAGE_NORMAL
);
742 static inline void rb_inc_page(struct ring_buffer_per_cpu
*cpu_buffer
,
743 struct buffer_page
**bpage
)
745 struct list_head
*p
= rb_list_head((*bpage
)->list
.next
);
747 *bpage
= list_entry(p
, struct buffer_page
, list
);
750 static struct buffer_page
*
751 rb_set_head_page(struct ring_buffer_per_cpu
*cpu_buffer
)
753 struct buffer_page
*head
;
754 struct buffer_page
*page
;
755 struct list_head
*list
;
758 if (RB_WARN_ON(cpu_buffer
, !cpu_buffer
->head_page
))
762 list
= cpu_buffer
->pages
;
763 if (RB_WARN_ON(cpu_buffer
, rb_list_head(list
->prev
->next
) != list
))
766 page
= head
= cpu_buffer
->head_page
;
768 * It is possible that the writer moves the header behind
769 * where we started, and we miss in one loop.
770 * A second loop should grab the header, but we'll do
771 * three loops just because I'm paranoid.
773 for (i
= 0; i
< 3; i
++) {
775 if (rb_is_head_page(cpu_buffer
, page
, page
->list
.prev
)) {
776 cpu_buffer
->head_page
= page
;
779 rb_inc_page(cpu_buffer
, &page
);
780 } while (page
!= head
);
783 RB_WARN_ON(cpu_buffer
, 1);
788 static int rb_head_page_replace(struct buffer_page
*old
,
789 struct buffer_page
*new)
791 unsigned long *ptr
= (unsigned long *)&old
->list
.prev
->next
;
795 val
= *ptr
& ~RB_FLAG_MASK
;
798 ret
= cmpxchg(ptr
, val
, (unsigned long)&new->list
);
804 * rb_tail_page_update - move the tail page forward
806 * Returns 1 if moved tail page, 0 if someone else did.
808 static int rb_tail_page_update(struct ring_buffer_per_cpu
*cpu_buffer
,
809 struct buffer_page
*tail_page
,
810 struct buffer_page
*next_page
)
812 struct buffer_page
*old_tail
;
813 unsigned long old_entries
;
814 unsigned long old_write
;
818 * The tail page now needs to be moved forward.
820 * We need to reset the tail page, but without messing
821 * with possible erasing of data brought in by interrupts
822 * that have moved the tail page and are currently on it.
824 * We add a counter to the write field to denote this.
826 old_write
= local_add_return(RB_WRITE_INTCNT
, &next_page
->write
);
827 old_entries
= local_add_return(RB_WRITE_INTCNT
, &next_page
->entries
);
830 * Just make sure we have seen our old_write and synchronize
831 * with any interrupts that come in.
836 * If the tail page is still the same as what we think
837 * it is, then it is up to us to update the tail
840 if (tail_page
== cpu_buffer
->tail_page
) {
841 /* Zero the write counter */
842 unsigned long val
= old_write
& ~RB_WRITE_MASK
;
843 unsigned long eval
= old_entries
& ~RB_WRITE_MASK
;
846 * This will only succeed if an interrupt did
847 * not come in and change it. In which case, we
848 * do not want to modify it.
850 * We add (void) to let the compiler know that we do not care
851 * about the return value of these functions. We use the
852 * cmpxchg to only update if an interrupt did not already
853 * do it for us. If the cmpxchg fails, we don't care.
855 (void)local_cmpxchg(&next_page
->write
, old_write
, val
);
856 (void)local_cmpxchg(&next_page
->entries
, old_entries
, eval
);
859 * No need to worry about races with clearing out the commit.
860 * it only can increment when a commit takes place. But that
861 * only happens in the outer most nested commit.
863 local_set(&next_page
->page
->commit
, 0);
865 old_tail
= cmpxchg(&cpu_buffer
->tail_page
,
866 tail_page
, next_page
);
868 if (old_tail
== tail_page
)
875 static int rb_check_bpage(struct ring_buffer_per_cpu
*cpu_buffer
,
876 struct buffer_page
*bpage
)
878 unsigned long val
= (unsigned long)bpage
;
880 if (RB_WARN_ON(cpu_buffer
, val
& RB_FLAG_MASK
))
887 * rb_check_list - make sure a pointer to a list has the last bits zero
889 static int rb_check_list(struct ring_buffer_per_cpu
*cpu_buffer
,
890 struct list_head
*list
)
892 if (RB_WARN_ON(cpu_buffer
, rb_list_head(list
->prev
) != list
->prev
))
894 if (RB_WARN_ON(cpu_buffer
, rb_list_head(list
->next
) != list
->next
))
900 * check_pages - integrity check of buffer pages
901 * @cpu_buffer: CPU buffer with pages to test
903 * As a safety measure we check to make sure the data pages have not
906 static int rb_check_pages(struct ring_buffer_per_cpu
*cpu_buffer
)
908 struct list_head
*head
= cpu_buffer
->pages
;
909 struct buffer_page
*bpage
, *tmp
;
911 rb_head_page_deactivate(cpu_buffer
);
913 if (RB_WARN_ON(cpu_buffer
, head
->next
->prev
!= head
))
915 if (RB_WARN_ON(cpu_buffer
, head
->prev
->next
!= head
))
918 if (rb_check_list(cpu_buffer
, head
))
921 list_for_each_entry_safe(bpage
, tmp
, head
, list
) {
922 if (RB_WARN_ON(cpu_buffer
,
923 bpage
->list
.next
->prev
!= &bpage
->list
))
925 if (RB_WARN_ON(cpu_buffer
,
926 bpage
->list
.prev
->next
!= &bpage
->list
))
928 if (rb_check_list(cpu_buffer
, &bpage
->list
))
932 rb_head_page_activate(cpu_buffer
);
937 static int rb_allocate_pages(struct ring_buffer_per_cpu
*cpu_buffer
,
940 struct buffer_page
*bpage
, *tmp
;
947 for (i
= 0; i
< nr_pages
; i
++) {
948 bpage
= kzalloc_node(ALIGN(sizeof(*bpage
), cache_line_size()),
949 GFP_KERNEL
, cpu_to_node(cpu_buffer
->cpu
));
953 rb_check_bpage(cpu_buffer
, bpage
);
955 list_add(&bpage
->list
, &pages
);
957 addr
= __get_free_page(GFP_KERNEL
);
960 bpage
->page
= (void *)addr
;
961 rb_init_page(bpage
->page
);
965 * The ring buffer page list is a circular list that does not
966 * start and end with a list head. All page list items point to
969 cpu_buffer
->pages
= pages
.next
;
972 rb_check_pages(cpu_buffer
);
977 list_for_each_entry_safe(bpage
, tmp
, &pages
, list
) {
978 list_del_init(&bpage
->list
);
979 free_buffer_page(bpage
);
984 static struct ring_buffer_per_cpu
*
985 rb_allocate_cpu_buffer(struct ring_buffer
*buffer
, int cpu
)
987 struct ring_buffer_per_cpu
*cpu_buffer
;
988 struct buffer_page
*bpage
;
992 cpu_buffer
= kzalloc_node(ALIGN(sizeof(*cpu_buffer
), cache_line_size()),
993 GFP_KERNEL
, cpu_to_node(cpu
));
997 cpu_buffer
->cpu
= cpu
;
998 cpu_buffer
->buffer
= buffer
;
999 spin_lock_init(&cpu_buffer
->reader_lock
);
1000 lockdep_set_class(&cpu_buffer
->reader_lock
, buffer
->reader_lock_key
);
1001 cpu_buffer
->lock
= (raw_spinlock_t
)__RAW_SPIN_LOCK_UNLOCKED
;
1003 bpage
= kzalloc_node(ALIGN(sizeof(*bpage
), cache_line_size()),
1004 GFP_KERNEL
, cpu_to_node(cpu
));
1006 goto fail_free_buffer
;
1008 rb_check_bpage(cpu_buffer
, bpage
);
1010 cpu_buffer
->reader_page
= bpage
;
1011 addr
= __get_free_page(GFP_KERNEL
);
1013 goto fail_free_reader
;
1014 bpage
->page
= (void *)addr
;
1015 rb_init_page(bpage
->page
);
1017 INIT_LIST_HEAD(&cpu_buffer
->reader_page
->list
);
1019 ret
= rb_allocate_pages(cpu_buffer
, buffer
->pages
);
1021 goto fail_free_reader
;
1023 cpu_buffer
->head_page
1024 = list_entry(cpu_buffer
->pages
, struct buffer_page
, list
);
1025 cpu_buffer
->tail_page
= cpu_buffer
->commit_page
= cpu_buffer
->head_page
;
1027 rb_head_page_activate(cpu_buffer
);
1032 free_buffer_page(cpu_buffer
->reader_page
);
1039 static void rb_free_cpu_buffer(struct ring_buffer_per_cpu
*cpu_buffer
)
1041 struct list_head
*head
= cpu_buffer
->pages
;
1042 struct buffer_page
*bpage
, *tmp
;
1044 free_buffer_page(cpu_buffer
->reader_page
);
1046 rb_head_page_deactivate(cpu_buffer
);
1049 list_for_each_entry_safe(bpage
, tmp
, head
, list
) {
1050 list_del_init(&bpage
->list
);
1051 free_buffer_page(bpage
);
1053 bpage
= list_entry(head
, struct buffer_page
, list
);
1054 free_buffer_page(bpage
);
1060 #ifdef CONFIG_HOTPLUG_CPU
1061 static int rb_cpu_notify(struct notifier_block
*self
,
1062 unsigned long action
, void *hcpu
);
1066 * ring_buffer_alloc - allocate a new ring_buffer
1067 * @size: the size in bytes per cpu that is needed.
1068 * @flags: attributes to set for the ring buffer.
1070 * Currently the only flag that is available is the RB_FL_OVERWRITE
1071 * flag. This flag means that the buffer will overwrite old data
1072 * when the buffer wraps. If this flag is not set, the buffer will
1073 * drop data when the tail hits the head.
1075 struct ring_buffer
*__ring_buffer_alloc(unsigned long size
, unsigned flags
,
1076 struct lock_class_key
*key
)
1078 struct ring_buffer
*buffer
;
1082 /* keep it in its own cache line */
1083 buffer
= kzalloc(ALIGN(sizeof(*buffer
), cache_line_size()),
1088 if (!alloc_cpumask_var(&buffer
->cpumask
, GFP_KERNEL
))
1089 goto fail_free_buffer
;
1091 buffer
->pages
= DIV_ROUND_UP(size
, BUF_PAGE_SIZE
);
1092 buffer
->flags
= flags
;
1093 buffer
->clock
= trace_clock_local
;
1094 buffer
->reader_lock_key
= key
;
1096 /* need at least two pages */
1097 if (buffer
->pages
< 2)
1101 * In case of non-hotplug cpu, if the ring-buffer is allocated
1102 * in early initcall, it will not be notified of secondary cpus.
1103 * In that off case, we need to allocate for all possible cpus.
1105 #ifdef CONFIG_HOTPLUG_CPU
1107 cpumask_copy(buffer
->cpumask
, cpu_online_mask
);
1109 cpumask_copy(buffer
->cpumask
, cpu_possible_mask
);
1111 buffer
->cpus
= nr_cpu_ids
;
1113 bsize
= sizeof(void *) * nr_cpu_ids
;
1114 buffer
->buffers
= kzalloc(ALIGN(bsize
, cache_line_size()),
1116 if (!buffer
->buffers
)
1117 goto fail_free_cpumask
;
1119 for_each_buffer_cpu(buffer
, cpu
) {
1120 buffer
->buffers
[cpu
] =
1121 rb_allocate_cpu_buffer(buffer
, cpu
);
1122 if (!buffer
->buffers
[cpu
])
1123 goto fail_free_buffers
;
1126 #ifdef CONFIG_HOTPLUG_CPU
1127 buffer
->cpu_notify
.notifier_call
= rb_cpu_notify
;
1128 buffer
->cpu_notify
.priority
= 0;
1129 register_cpu_notifier(&buffer
->cpu_notify
);
1133 mutex_init(&buffer
->mutex
);
1138 for_each_buffer_cpu(buffer
, cpu
) {
1139 if (buffer
->buffers
[cpu
])
1140 rb_free_cpu_buffer(buffer
->buffers
[cpu
]);
1142 kfree(buffer
->buffers
);
1145 free_cpumask_var(buffer
->cpumask
);
1152 EXPORT_SYMBOL_GPL(__ring_buffer_alloc
);
1155 * ring_buffer_free - free a ring buffer.
1156 * @buffer: the buffer to free.
1159 ring_buffer_free(struct ring_buffer
*buffer
)
1165 #ifdef CONFIG_HOTPLUG_CPU
1166 unregister_cpu_notifier(&buffer
->cpu_notify
);
1169 for_each_buffer_cpu(buffer
, cpu
)
1170 rb_free_cpu_buffer(buffer
->buffers
[cpu
]);
1174 kfree(buffer
->buffers
);
1175 free_cpumask_var(buffer
->cpumask
);
1179 EXPORT_SYMBOL_GPL(ring_buffer_free
);
1181 void ring_buffer_set_clock(struct ring_buffer
*buffer
,
1184 buffer
->clock
= clock
;
1187 static void rb_reset_cpu(struct ring_buffer_per_cpu
*cpu_buffer
);
1190 rb_remove_pages(struct ring_buffer_per_cpu
*cpu_buffer
, unsigned nr_pages
)
1192 struct buffer_page
*bpage
;
1193 struct list_head
*p
;
1196 atomic_inc(&cpu_buffer
->record_disabled
);
1197 synchronize_sched();
1199 rb_head_page_deactivate(cpu_buffer
);
1201 for (i
= 0; i
< nr_pages
; i
++) {
1202 if (RB_WARN_ON(cpu_buffer
, list_empty(cpu_buffer
->pages
)))
1204 p
= cpu_buffer
->pages
->next
;
1205 bpage
= list_entry(p
, struct buffer_page
, list
);
1206 list_del_init(&bpage
->list
);
1207 free_buffer_page(bpage
);
1209 if (RB_WARN_ON(cpu_buffer
, list_empty(cpu_buffer
->pages
)))
1212 rb_reset_cpu(cpu_buffer
);
1214 rb_check_pages(cpu_buffer
);
1216 atomic_dec(&cpu_buffer
->record_disabled
);
1221 rb_insert_pages(struct ring_buffer_per_cpu
*cpu_buffer
,
1222 struct list_head
*pages
, unsigned nr_pages
)
1224 struct buffer_page
*bpage
;
1225 struct list_head
*p
;
1228 atomic_inc(&cpu_buffer
->record_disabled
);
1229 synchronize_sched();
1231 spin_lock_irq(&cpu_buffer
->reader_lock
);
1232 rb_head_page_deactivate(cpu_buffer
);
1234 for (i
= 0; i
< nr_pages
; i
++) {
1235 if (RB_WARN_ON(cpu_buffer
, list_empty(pages
)))
1238 bpage
= list_entry(p
, struct buffer_page
, list
);
1239 list_del_init(&bpage
->list
);
1240 list_add_tail(&bpage
->list
, cpu_buffer
->pages
);
1242 rb_reset_cpu(cpu_buffer
);
1243 spin_unlock_irq(&cpu_buffer
->reader_lock
);
1245 rb_check_pages(cpu_buffer
);
1247 atomic_dec(&cpu_buffer
->record_disabled
);
1251 * ring_buffer_resize - resize the ring buffer
1252 * @buffer: the buffer to resize.
1253 * @size: the new size.
1255 * The tracer is responsible for making sure that the buffer is
1256 * not being used while changing the size.
1257 * Note: We may be able to change the above requirement by using
1258 * RCU synchronizations.
1260 * Minimum size is 2 * BUF_PAGE_SIZE.
1262 * Returns -1 on failure.
1264 int ring_buffer_resize(struct ring_buffer
*buffer
, unsigned long size
)
1266 struct ring_buffer_per_cpu
*cpu_buffer
;
1267 unsigned nr_pages
, rm_pages
, new_pages
;
1268 struct buffer_page
*bpage
, *tmp
;
1269 unsigned long buffer_size
;
1275 * Always succeed at resizing a non-existent buffer:
1280 size
= DIV_ROUND_UP(size
, BUF_PAGE_SIZE
);
1281 size
*= BUF_PAGE_SIZE
;
1282 buffer_size
= buffer
->pages
* BUF_PAGE_SIZE
;
1284 /* we need a minimum of two pages */
1285 if (size
< BUF_PAGE_SIZE
* 2)
1286 size
= BUF_PAGE_SIZE
* 2;
1288 if (size
== buffer_size
)
1291 mutex_lock(&buffer
->mutex
);
1294 nr_pages
= DIV_ROUND_UP(size
, BUF_PAGE_SIZE
);
1296 if (size
< buffer_size
) {
1298 /* easy case, just free pages */
1299 if (RB_WARN_ON(buffer
, nr_pages
>= buffer
->pages
))
1302 rm_pages
= buffer
->pages
- nr_pages
;
1304 for_each_buffer_cpu(buffer
, cpu
) {
1305 cpu_buffer
= buffer
->buffers
[cpu
];
1306 rb_remove_pages(cpu_buffer
, rm_pages
);
1312 * This is a bit more difficult. We only want to add pages
1313 * when we can allocate enough for all CPUs. We do this
1314 * by allocating all the pages and storing them on a local
1315 * link list. If we succeed in our allocation, then we
1316 * add these pages to the cpu_buffers. Otherwise we just free
1317 * them all and return -ENOMEM;
1319 if (RB_WARN_ON(buffer
, nr_pages
<= buffer
->pages
))
1322 new_pages
= nr_pages
- buffer
->pages
;
1324 for_each_buffer_cpu(buffer
, cpu
) {
1325 for (i
= 0; i
< new_pages
; i
++) {
1326 bpage
= kzalloc_node(ALIGN(sizeof(*bpage
),
1328 GFP_KERNEL
, cpu_to_node(cpu
));
1331 list_add(&bpage
->list
, &pages
);
1332 addr
= __get_free_page(GFP_KERNEL
);
1335 bpage
->page
= (void *)addr
;
1336 rb_init_page(bpage
->page
);
1340 for_each_buffer_cpu(buffer
, cpu
) {
1341 cpu_buffer
= buffer
->buffers
[cpu
];
1342 rb_insert_pages(cpu_buffer
, &pages
, new_pages
);
1345 if (RB_WARN_ON(buffer
, !list_empty(&pages
)))
1349 buffer
->pages
= nr_pages
;
1351 mutex_unlock(&buffer
->mutex
);
1356 list_for_each_entry_safe(bpage
, tmp
, &pages
, list
) {
1357 list_del_init(&bpage
->list
);
1358 free_buffer_page(bpage
);
1361 mutex_unlock(&buffer
->mutex
);
1365 * Something went totally wrong, and we are too paranoid
1366 * to even clean up the mess.
1370 mutex_unlock(&buffer
->mutex
);
1373 EXPORT_SYMBOL_GPL(ring_buffer_resize
);
1375 static inline void *
1376 __rb_data_page_index(struct buffer_data_page
*bpage
, unsigned index
)
1378 return bpage
->data
+ index
;
1381 static inline void *__rb_page_index(struct buffer_page
*bpage
, unsigned index
)
1383 return bpage
->page
->data
+ index
;
1386 static inline struct ring_buffer_event
*
1387 rb_reader_event(struct ring_buffer_per_cpu
*cpu_buffer
)
1389 return __rb_page_index(cpu_buffer
->reader_page
,
1390 cpu_buffer
->reader_page
->read
);
1393 static inline struct ring_buffer_event
*
1394 rb_iter_head_event(struct ring_buffer_iter
*iter
)
1396 return __rb_page_index(iter
->head_page
, iter
->head
);
1399 static inline unsigned long rb_page_write(struct buffer_page
*bpage
)
1401 return local_read(&bpage
->write
) & RB_WRITE_MASK
;
1404 static inline unsigned rb_page_commit(struct buffer_page
*bpage
)
1406 return local_read(&bpage
->page
->commit
);
1409 static inline unsigned long rb_page_entries(struct buffer_page
*bpage
)
1411 return local_read(&bpage
->entries
) & RB_WRITE_MASK
;
1414 /* Size is determined by what has been commited */
1415 static inline unsigned rb_page_size(struct buffer_page
*bpage
)
1417 return rb_page_commit(bpage
);
1420 static inline unsigned
1421 rb_commit_index(struct ring_buffer_per_cpu
*cpu_buffer
)
1423 return rb_page_commit(cpu_buffer
->commit_page
);
1426 static inline unsigned
1427 rb_event_index(struct ring_buffer_event
*event
)
1429 unsigned long addr
= (unsigned long)event
;
1431 return (addr
& ~PAGE_MASK
) - BUF_PAGE_HDR_SIZE
;
1435 rb_event_is_commit(struct ring_buffer_per_cpu
*cpu_buffer
,
1436 struct ring_buffer_event
*event
)
1438 unsigned long addr
= (unsigned long)event
;
1439 unsigned long index
;
1441 index
= rb_event_index(event
);
1444 return cpu_buffer
->commit_page
->page
== (void *)addr
&&
1445 rb_commit_index(cpu_buffer
) == index
;
1449 rb_set_commit_to_write(struct ring_buffer_per_cpu
*cpu_buffer
)
1451 unsigned long max_count
;
1454 * We only race with interrupts and NMIs on this CPU.
1455 * If we own the commit event, then we can commit
1456 * all others that interrupted us, since the interruptions
1457 * are in stack format (they finish before they come
1458 * back to us). This allows us to do a simple loop to
1459 * assign the commit to the tail.
1462 max_count
= cpu_buffer
->buffer
->pages
* 100;
1464 while (cpu_buffer
->commit_page
!= cpu_buffer
->tail_page
) {
1465 if (RB_WARN_ON(cpu_buffer
, !(--max_count
)))
1467 if (RB_WARN_ON(cpu_buffer
,
1468 rb_is_reader_page(cpu_buffer
->tail_page
)))
1470 local_set(&cpu_buffer
->commit_page
->page
->commit
,
1471 rb_page_write(cpu_buffer
->commit_page
));
1472 rb_inc_page(cpu_buffer
, &cpu_buffer
->commit_page
);
1473 cpu_buffer
->write_stamp
=
1474 cpu_buffer
->commit_page
->page
->time_stamp
;
1475 /* add barrier to keep gcc from optimizing too much */
1478 while (rb_commit_index(cpu_buffer
) !=
1479 rb_page_write(cpu_buffer
->commit_page
)) {
1481 local_set(&cpu_buffer
->commit_page
->page
->commit
,
1482 rb_page_write(cpu_buffer
->commit_page
));
1483 RB_WARN_ON(cpu_buffer
,
1484 local_read(&cpu_buffer
->commit_page
->page
->commit
) &
1489 /* again, keep gcc from optimizing */
1493 * If an interrupt came in just after the first while loop
1494 * and pushed the tail page forward, we will be left with
1495 * a dangling commit that will never go forward.
1497 if (unlikely(cpu_buffer
->commit_page
!= cpu_buffer
->tail_page
))
1501 static void rb_reset_reader_page(struct ring_buffer_per_cpu
*cpu_buffer
)
1503 cpu_buffer
->read_stamp
= cpu_buffer
->reader_page
->page
->time_stamp
;
1504 cpu_buffer
->reader_page
->read
= 0;
1507 static void rb_inc_iter(struct ring_buffer_iter
*iter
)
1509 struct ring_buffer_per_cpu
*cpu_buffer
= iter
->cpu_buffer
;
1512 * The iterator could be on the reader page (it starts there).
1513 * But the head could have moved, since the reader was
1514 * found. Check for this case and assign the iterator
1515 * to the head page instead of next.
1517 if (iter
->head_page
== cpu_buffer
->reader_page
)
1518 iter
->head_page
= rb_set_head_page(cpu_buffer
);
1520 rb_inc_page(cpu_buffer
, &iter
->head_page
);
1522 iter
->read_stamp
= iter
->head_page
->page
->time_stamp
;
1527 * ring_buffer_update_event - update event type and data
1528 * @event: the even to update
1529 * @type: the type of event
1530 * @length: the size of the event field in the ring buffer
1532 * Update the type and data fields of the event. The length
1533 * is the actual size that is written to the ring buffer,
1534 * and with this, we can determine what to place into the
1538 rb_update_event(struct ring_buffer_event
*event
,
1539 unsigned type
, unsigned length
)
1541 event
->type_len
= type
;
1545 case RINGBUF_TYPE_PADDING
:
1546 case RINGBUF_TYPE_TIME_EXTEND
:
1547 case RINGBUF_TYPE_TIME_STAMP
:
1551 length
-= RB_EVNT_HDR_SIZE
;
1552 if (length
> RB_MAX_SMALL_DATA
)
1553 event
->array
[0] = length
;
1555 event
->type_len
= DIV_ROUND_UP(length
, RB_ALIGNMENT
);
1563 * rb_handle_head_page - writer hit the head page
1565 * Returns: +1 to retry page
1570 rb_handle_head_page(struct ring_buffer_per_cpu
*cpu_buffer
,
1571 struct buffer_page
*tail_page
,
1572 struct buffer_page
*next_page
)
1574 struct buffer_page
*new_head
;
1579 entries
= rb_page_entries(next_page
);
1582 * The hard part is here. We need to move the head
1583 * forward, and protect against both readers on
1584 * other CPUs and writers coming in via interrupts.
1586 type
= rb_head_page_set_update(cpu_buffer
, next_page
, tail_page
,
1590 * type can be one of four:
1591 * NORMAL - an interrupt already moved it for us
1592 * HEAD - we are the first to get here.
1593 * UPDATE - we are the interrupt interrupting
1595 * MOVED - a reader on another CPU moved the next
1596 * pointer to its reader page. Give up
1603 * We changed the head to UPDATE, thus
1604 * it is our responsibility to update
1607 local_add(entries
, &cpu_buffer
->overrun
);
1610 * The entries will be zeroed out when we move the
1614 /* still more to do */
1617 case RB_PAGE_UPDATE
:
1619 * This is an interrupt that interrupt the
1620 * previous update. Still more to do.
1623 case RB_PAGE_NORMAL
:
1625 * An interrupt came in before the update
1626 * and processed this for us.
1627 * Nothing left to do.
1632 * The reader is on another CPU and just did
1633 * a swap with our next_page.
1638 RB_WARN_ON(cpu_buffer
, 1); /* WTF??? */
1643 * Now that we are here, the old head pointer is
1644 * set to UPDATE. This will keep the reader from
1645 * swapping the head page with the reader page.
1646 * The reader (on another CPU) will spin till
1649 * We just need to protect against interrupts
1650 * doing the job. We will set the next pointer
1651 * to HEAD. After that, we set the old pointer
1652 * to NORMAL, but only if it was HEAD before.
1653 * otherwise we are an interrupt, and only
1654 * want the outer most commit to reset it.
1656 new_head
= next_page
;
1657 rb_inc_page(cpu_buffer
, &new_head
);
1659 ret
= rb_head_page_set_head(cpu_buffer
, new_head
, next_page
,
1663 * Valid returns are:
1664 * HEAD - an interrupt came in and already set it.
1665 * NORMAL - One of two things:
1666 * 1) We really set it.
1667 * 2) A bunch of interrupts came in and moved
1668 * the page forward again.
1672 case RB_PAGE_NORMAL
:
1676 RB_WARN_ON(cpu_buffer
, 1);
1681 * It is possible that an interrupt came in,
1682 * set the head up, then more interrupts came in
1683 * and moved it again. When we get back here,
1684 * the page would have been set to NORMAL but we
1685 * just set it back to HEAD.
1687 * How do you detect this? Well, if that happened
1688 * the tail page would have moved.
1690 if (ret
== RB_PAGE_NORMAL
) {
1692 * If the tail had moved passed next, then we need
1693 * to reset the pointer.
1695 if (cpu_buffer
->tail_page
!= tail_page
&&
1696 cpu_buffer
->tail_page
!= next_page
)
1697 rb_head_page_set_normal(cpu_buffer
, new_head
,
1703 * If this was the outer most commit (the one that
1704 * changed the original pointer from HEAD to UPDATE),
1705 * then it is up to us to reset it to NORMAL.
1707 if (type
== RB_PAGE_HEAD
) {
1708 ret
= rb_head_page_set_normal(cpu_buffer
, next_page
,
1711 if (RB_WARN_ON(cpu_buffer
,
1712 ret
!= RB_PAGE_UPDATE
))
1719 static unsigned rb_calculate_event_length(unsigned length
)
1721 struct ring_buffer_event event
; /* Used only for sizeof array */
1723 /* zero length can cause confusions */
1727 if (length
> RB_MAX_SMALL_DATA
)
1728 length
+= sizeof(event
.array
[0]);
1730 length
+= RB_EVNT_HDR_SIZE
;
1731 length
= ALIGN(length
, RB_ALIGNMENT
);
1737 rb_reset_tail(struct ring_buffer_per_cpu
*cpu_buffer
,
1738 struct buffer_page
*tail_page
,
1739 unsigned long tail
, unsigned long length
)
1741 struct ring_buffer_event
*event
;
1744 * Only the event that crossed the page boundary
1745 * must fill the old tail_page with padding.
1747 if (tail
>= BUF_PAGE_SIZE
) {
1748 local_sub(length
, &tail_page
->write
);
1752 event
= __rb_page_index(tail_page
, tail
);
1753 kmemcheck_annotate_bitfield(event
, bitfield
);
1756 * If this event is bigger than the minimum size, then
1757 * we need to be careful that we don't subtract the
1758 * write counter enough to allow another writer to slip
1760 * We put in a discarded commit instead, to make sure
1761 * that this space is not used again.
1763 * If we are less than the minimum size, we don't need to
1766 if (tail
> (BUF_PAGE_SIZE
- RB_EVNT_MIN_SIZE
)) {
1767 /* No room for any events */
1769 /* Mark the rest of the page with padding */
1770 rb_event_set_padding(event
);
1772 /* Set the write back to the previous setting */
1773 local_sub(length
, &tail_page
->write
);
1777 /* Put in a discarded event */
1778 event
->array
[0] = (BUF_PAGE_SIZE
- tail
) - RB_EVNT_HDR_SIZE
;
1779 event
->type_len
= RINGBUF_TYPE_PADDING
;
1780 /* time delta must be non zero */
1781 event
->time_delta
= 1;
1783 /* Set write to end of buffer */
1784 length
= (tail
+ length
) - BUF_PAGE_SIZE
;
1785 local_sub(length
, &tail_page
->write
);
1788 static struct ring_buffer_event
*
1789 rb_move_tail(struct ring_buffer_per_cpu
*cpu_buffer
,
1790 unsigned long length
, unsigned long tail
,
1791 struct buffer_page
*commit_page
,
1792 struct buffer_page
*tail_page
, u64
*ts
)
1794 struct ring_buffer
*buffer
= cpu_buffer
->buffer
;
1795 struct buffer_page
*next_page
;
1798 next_page
= tail_page
;
1800 rb_inc_page(cpu_buffer
, &next_page
);
1803 * If for some reason, we had an interrupt storm that made
1804 * it all the way around the buffer, bail, and warn
1807 if (unlikely(next_page
== commit_page
)) {
1808 local_inc(&cpu_buffer
->commit_overrun
);
1813 * This is where the fun begins!
1815 * We are fighting against races between a reader that
1816 * could be on another CPU trying to swap its reader
1817 * page with the buffer head.
1819 * We are also fighting against interrupts coming in and
1820 * moving the head or tail on us as well.
1822 * If the next page is the head page then we have filled
1823 * the buffer, unless the commit page is still on the
1826 if (rb_is_head_page(cpu_buffer
, next_page
, &tail_page
->list
)) {
1829 * If the commit is not on the reader page, then
1830 * move the header page.
1832 if (!rb_is_reader_page(cpu_buffer
->commit_page
)) {
1834 * If we are not in overwrite mode,
1835 * this is easy, just stop here.
1837 if (!(buffer
->flags
& RB_FL_OVERWRITE
))
1840 ret
= rb_handle_head_page(cpu_buffer
,
1849 * We need to be careful here too. The
1850 * commit page could still be on the reader
1851 * page. We could have a small buffer, and
1852 * have filled up the buffer with events
1853 * from interrupts and such, and wrapped.
1855 * Note, if the tail page is also the on the
1856 * reader_page, we let it move out.
1858 if (unlikely((cpu_buffer
->commit_page
!=
1859 cpu_buffer
->tail_page
) &&
1860 (cpu_buffer
->commit_page
==
1861 cpu_buffer
->reader_page
))) {
1862 local_inc(&cpu_buffer
->commit_overrun
);
1868 ret
= rb_tail_page_update(cpu_buffer
, tail_page
, next_page
);
1871 * Nested commits always have zero deltas, so
1872 * just reread the time stamp
1874 *ts
= rb_time_stamp(buffer
, cpu_buffer
->cpu
);
1875 next_page
->page
->time_stamp
= *ts
;
1880 rb_reset_tail(cpu_buffer
, tail_page
, tail
, length
);
1882 /* fail and let the caller try again */
1883 return ERR_PTR(-EAGAIN
);
1887 rb_reset_tail(cpu_buffer
, tail_page
, tail
, length
);
1892 static struct ring_buffer_event
*
1893 __rb_reserve_next(struct ring_buffer_per_cpu
*cpu_buffer
,
1894 unsigned type
, unsigned long length
, u64
*ts
)
1896 struct buffer_page
*tail_page
, *commit_page
;
1897 struct ring_buffer_event
*event
;
1898 unsigned long tail
, write
;
1900 commit_page
= cpu_buffer
->commit_page
;
1901 /* we just need to protect against interrupts */
1903 tail_page
= cpu_buffer
->tail_page
;
1904 write
= local_add_return(length
, &tail_page
->write
);
1906 /* set write to only the index of the write */
1907 write
&= RB_WRITE_MASK
;
1908 tail
= write
- length
;
1910 /* See if we shot pass the end of this buffer page */
1911 if (write
> BUF_PAGE_SIZE
)
1912 return rb_move_tail(cpu_buffer
, length
, tail
,
1913 commit_page
, tail_page
, ts
);
1915 /* We reserved something on the buffer */
1917 event
= __rb_page_index(tail_page
, tail
);
1918 kmemcheck_annotate_bitfield(event
, bitfield
);
1919 rb_update_event(event
, type
, length
);
1921 /* The passed in type is zero for DATA */
1923 local_inc(&tail_page
->entries
);
1926 * If this is the first commit on the page, then update
1930 tail_page
->page
->time_stamp
= *ts
;
1936 rb_try_to_discard(struct ring_buffer_per_cpu
*cpu_buffer
,
1937 struct ring_buffer_event
*event
)
1939 unsigned long new_index
, old_index
;
1940 struct buffer_page
*bpage
;
1941 unsigned long index
;
1944 new_index
= rb_event_index(event
);
1945 old_index
= new_index
+ rb_event_length(event
);
1946 addr
= (unsigned long)event
;
1949 bpage
= cpu_buffer
->tail_page
;
1951 if (bpage
->page
== (void *)addr
&& rb_page_write(bpage
) == old_index
) {
1952 unsigned long write_mask
=
1953 local_read(&bpage
->write
) & ~RB_WRITE_MASK
;
1955 * This is on the tail page. It is possible that
1956 * a write could come in and move the tail page
1957 * and write to the next page. That is fine
1958 * because we just shorten what is on this page.
1960 old_index
+= write_mask
;
1961 new_index
+= write_mask
;
1962 index
= local_cmpxchg(&bpage
->write
, old_index
, new_index
);
1963 if (index
== old_index
)
1967 /* could not discard */
1972 rb_add_time_stamp(struct ring_buffer_per_cpu
*cpu_buffer
,
1973 u64
*ts
, u64
*delta
)
1975 struct ring_buffer_event
*event
;
1979 if (unlikely(*delta
> (1ULL << 59) && !once
++)) {
1980 printk(KERN_WARNING
"Delta way too big! %llu"
1981 " ts=%llu write stamp = %llu\n",
1982 (unsigned long long)*delta
,
1983 (unsigned long long)*ts
,
1984 (unsigned long long)cpu_buffer
->write_stamp
);
1989 * The delta is too big, we to add a
1992 event
= __rb_reserve_next(cpu_buffer
,
1993 RINGBUF_TYPE_TIME_EXTEND
,
1999 if (PTR_ERR(event
) == -EAGAIN
)
2002 /* Only a commited time event can update the write stamp */
2003 if (rb_event_is_commit(cpu_buffer
, event
)) {
2005 * If this is the first on the page, then it was
2006 * updated with the page itself. Try to discard it
2007 * and if we can't just make it zero.
2009 if (rb_event_index(event
)) {
2010 event
->time_delta
= *delta
& TS_MASK
;
2011 event
->array
[0] = *delta
>> TS_SHIFT
;
2013 /* try to discard, since we do not need this */
2014 if (!rb_try_to_discard(cpu_buffer
, event
)) {
2015 /* nope, just zero it */
2016 event
->time_delta
= 0;
2017 event
->array
[0] = 0;
2020 cpu_buffer
->write_stamp
= *ts
;
2021 /* let the caller know this was the commit */
2024 /* Try to discard the event */
2025 if (!rb_try_to_discard(cpu_buffer
, event
)) {
2026 /* Darn, this is just wasted space */
2027 event
->time_delta
= 0;
2028 event
->array
[0] = 0;
2038 static void rb_start_commit(struct ring_buffer_per_cpu
*cpu_buffer
)
2040 local_inc(&cpu_buffer
->committing
);
2041 local_inc(&cpu_buffer
->commits
);
2044 static void rb_end_commit(struct ring_buffer_per_cpu
*cpu_buffer
)
2046 unsigned long commits
;
2048 if (RB_WARN_ON(cpu_buffer
,
2049 !local_read(&cpu_buffer
->committing
)))
2053 commits
= local_read(&cpu_buffer
->commits
);
2054 /* synchronize with interrupts */
2056 if (local_read(&cpu_buffer
->committing
) == 1)
2057 rb_set_commit_to_write(cpu_buffer
);
2059 local_dec(&cpu_buffer
->committing
);
2061 /* synchronize with interrupts */
2065 * Need to account for interrupts coming in between the
2066 * updating of the commit page and the clearing of the
2067 * committing counter.
2069 if (unlikely(local_read(&cpu_buffer
->commits
) != commits
) &&
2070 !local_read(&cpu_buffer
->committing
)) {
2071 local_inc(&cpu_buffer
->committing
);
2076 static struct ring_buffer_event
*
2077 rb_reserve_next_event(struct ring_buffer
*buffer
,
2078 struct ring_buffer_per_cpu
*cpu_buffer
,
2079 unsigned long length
)
2081 struct ring_buffer_event
*event
;
2086 rb_start_commit(cpu_buffer
);
2088 #ifdef CONFIG_RING_BUFFER_ALLOW_SWAP
2090 * Due to the ability to swap a cpu buffer from a buffer
2091 * it is possible it was swapped before we committed.
2092 * (committing stops a swap). We check for it here and
2093 * if it happened, we have to fail the write.
2096 if (unlikely(ACCESS_ONCE(cpu_buffer
->buffer
) != buffer
)) {
2097 local_dec(&cpu_buffer
->committing
);
2098 local_dec(&cpu_buffer
->commits
);
2103 length
= rb_calculate_event_length(length
);
2106 * We allow for interrupts to reenter here and do a trace.
2107 * If one does, it will cause this original code to loop
2108 * back here. Even with heavy interrupts happening, this
2109 * should only happen a few times in a row. If this happens
2110 * 1000 times in a row, there must be either an interrupt
2111 * storm or we have something buggy.
2114 if (RB_WARN_ON(cpu_buffer
, ++nr_loops
> 1000))
2117 ts
= rb_time_stamp(cpu_buffer
->buffer
, cpu_buffer
->cpu
);
2120 * Only the first commit can update the timestamp.
2121 * Yes there is a race here. If an interrupt comes in
2122 * just after the conditional and it traces too, then it
2123 * will also check the deltas. More than one timestamp may
2124 * also be made. But only the entry that did the actual
2125 * commit will be something other than zero.
2127 if (likely(cpu_buffer
->tail_page
== cpu_buffer
->commit_page
&&
2128 rb_page_write(cpu_buffer
->tail_page
) ==
2129 rb_commit_index(cpu_buffer
))) {
2132 diff
= ts
- cpu_buffer
->write_stamp
;
2134 /* make sure this diff is calculated here */
2137 /* Did the write stamp get updated already? */
2138 if (unlikely(ts
< cpu_buffer
->write_stamp
))
2142 if (unlikely(test_time_stamp(delta
))) {
2144 commit
= rb_add_time_stamp(cpu_buffer
, &ts
, &delta
);
2145 if (commit
== -EBUSY
)
2148 if (commit
== -EAGAIN
)
2151 RB_WARN_ON(cpu_buffer
, commit
< 0);
2156 event
= __rb_reserve_next(cpu_buffer
, 0, length
, &ts
);
2157 if (unlikely(PTR_ERR(event
) == -EAGAIN
))
2163 if (!rb_event_is_commit(cpu_buffer
, event
))
2166 event
->time_delta
= delta
;
2171 rb_end_commit(cpu_buffer
);
2175 #ifdef CONFIG_TRACING
2177 #define TRACE_RECURSIVE_DEPTH 16
2179 static int trace_recursive_lock(void)
2181 current
->trace_recursion
++;
2183 if (likely(current
->trace_recursion
< TRACE_RECURSIVE_DEPTH
))
2186 /* Disable all tracing before we do anything else */
2187 tracing_off_permanent();
2189 printk_once(KERN_WARNING
"Tracing recursion: depth[%ld]:"
2190 "HC[%lu]:SC[%lu]:NMI[%lu]\n",
2191 current
->trace_recursion
,
2192 hardirq_count() >> HARDIRQ_SHIFT
,
2193 softirq_count() >> SOFTIRQ_SHIFT
,
2200 static void trace_recursive_unlock(void)
2202 WARN_ON_ONCE(!current
->trace_recursion
);
2204 current
->trace_recursion
--;
2209 #define trace_recursive_lock() (0)
2210 #define trace_recursive_unlock() do { } while (0)
2214 static DEFINE_PER_CPU(int, rb_need_resched
);
2217 * ring_buffer_lock_reserve - reserve a part of the buffer
2218 * @buffer: the ring buffer to reserve from
2219 * @length: the length of the data to reserve (excluding event header)
2221 * Returns a reseverd event on the ring buffer to copy directly to.
2222 * The user of this interface will need to get the body to write into
2223 * and can use the ring_buffer_event_data() interface.
2225 * The length is the length of the data needed, not the event length
2226 * which also includes the event header.
2228 * Must be paired with ring_buffer_unlock_commit, unless NULL is returned.
2229 * If NULL is returned, then nothing has been allocated or locked.
2231 struct ring_buffer_event
*
2232 ring_buffer_lock_reserve(struct ring_buffer
*buffer
, unsigned long length
)
2234 struct ring_buffer_per_cpu
*cpu_buffer
;
2235 struct ring_buffer_event
*event
;
2238 if (ring_buffer_flags
!= RB_BUFFERS_ON
)
2241 if (atomic_read(&buffer
->record_disabled
))
2244 /* If we are tracing schedule, we don't want to recurse */
2245 resched
= ftrace_preempt_disable();
2247 if (trace_recursive_lock())
2250 cpu
= raw_smp_processor_id();
2252 if (!cpumask_test_cpu(cpu
, buffer
->cpumask
))
2255 cpu_buffer
= buffer
->buffers
[cpu
];
2257 if (atomic_read(&cpu_buffer
->record_disabled
))
2260 if (length
> BUF_MAX_DATA_SIZE
)
2263 event
= rb_reserve_next_event(buffer
, cpu_buffer
, length
);
2268 * Need to store resched state on this cpu.
2269 * Only the first needs to.
2272 if (preempt_count() == 1)
2273 per_cpu(rb_need_resched
, cpu
) = resched
;
2278 trace_recursive_unlock();
2281 ftrace_preempt_enable(resched
);
2284 EXPORT_SYMBOL_GPL(ring_buffer_lock_reserve
);
2287 rb_update_write_stamp(struct ring_buffer_per_cpu
*cpu_buffer
,
2288 struct ring_buffer_event
*event
)
2291 * The event first in the commit queue updates the
2294 if (rb_event_is_commit(cpu_buffer
, event
))
2295 cpu_buffer
->write_stamp
+= event
->time_delta
;
2298 static void rb_commit(struct ring_buffer_per_cpu
*cpu_buffer
,
2299 struct ring_buffer_event
*event
)
2301 local_inc(&cpu_buffer
->entries
);
2302 rb_update_write_stamp(cpu_buffer
, event
);
2303 rb_end_commit(cpu_buffer
);
2307 * ring_buffer_unlock_commit - commit a reserved
2308 * @buffer: The buffer to commit to
2309 * @event: The event pointer to commit.
2311 * This commits the data to the ring buffer, and releases any locks held.
2313 * Must be paired with ring_buffer_lock_reserve.
2315 int ring_buffer_unlock_commit(struct ring_buffer
*buffer
,
2316 struct ring_buffer_event
*event
)
2318 struct ring_buffer_per_cpu
*cpu_buffer
;
2319 int cpu
= raw_smp_processor_id();
2321 cpu_buffer
= buffer
->buffers
[cpu
];
2323 rb_commit(cpu_buffer
, event
);
2325 trace_recursive_unlock();
2328 * Only the last preempt count needs to restore preemption.
2330 if (preempt_count() == 1)
2331 ftrace_preempt_enable(per_cpu(rb_need_resched
, cpu
));
2333 preempt_enable_no_resched_notrace();
2337 EXPORT_SYMBOL_GPL(ring_buffer_unlock_commit
);
2339 static inline void rb_event_discard(struct ring_buffer_event
*event
)
2341 /* array[0] holds the actual length for the discarded event */
2342 event
->array
[0] = rb_event_data_length(event
) - RB_EVNT_HDR_SIZE
;
2343 event
->type_len
= RINGBUF_TYPE_PADDING
;
2344 /* time delta must be non zero */
2345 if (!event
->time_delta
)
2346 event
->time_delta
= 1;
2350 * Decrement the entries to the page that an event is on.
2351 * The event does not even need to exist, only the pointer
2352 * to the page it is on. This may only be called before the commit
2356 rb_decrement_entry(struct ring_buffer_per_cpu
*cpu_buffer
,
2357 struct ring_buffer_event
*event
)
2359 unsigned long addr
= (unsigned long)event
;
2360 struct buffer_page
*bpage
= cpu_buffer
->commit_page
;
2361 struct buffer_page
*start
;
2365 /* Do the likely case first */
2366 if (likely(bpage
->page
== (void *)addr
)) {
2367 local_dec(&bpage
->entries
);
2372 * Because the commit page may be on the reader page we
2373 * start with the next page and check the end loop there.
2375 rb_inc_page(cpu_buffer
, &bpage
);
2378 if (bpage
->page
== (void *)addr
) {
2379 local_dec(&bpage
->entries
);
2382 rb_inc_page(cpu_buffer
, &bpage
);
2383 } while (bpage
!= start
);
2385 /* commit not part of this buffer?? */
2386 RB_WARN_ON(cpu_buffer
, 1);
2390 * ring_buffer_commit_discard - discard an event that has not been committed
2391 * @buffer: the ring buffer
2392 * @event: non committed event to discard
2394 * Sometimes an event that is in the ring buffer needs to be ignored.
2395 * This function lets the user discard an event in the ring buffer
2396 * and then that event will not be read later.
2398 * This function only works if it is called before the the item has been
2399 * committed. It will try to free the event from the ring buffer
2400 * if another event has not been added behind it.
2402 * If another event has been added behind it, it will set the event
2403 * up as discarded, and perform the commit.
2405 * If this function is called, do not call ring_buffer_unlock_commit on
2408 void ring_buffer_discard_commit(struct ring_buffer
*buffer
,
2409 struct ring_buffer_event
*event
)
2411 struct ring_buffer_per_cpu
*cpu_buffer
;
2414 /* The event is discarded regardless */
2415 rb_event_discard(event
);
2417 cpu
= smp_processor_id();
2418 cpu_buffer
= buffer
->buffers
[cpu
];
2421 * This must only be called if the event has not been
2422 * committed yet. Thus we can assume that preemption
2423 * is still disabled.
2425 RB_WARN_ON(buffer
, !local_read(&cpu_buffer
->committing
));
2427 rb_decrement_entry(cpu_buffer
, event
);
2428 if (rb_try_to_discard(cpu_buffer
, event
))
2432 * The commit is still visible by the reader, so we
2433 * must still update the timestamp.
2435 rb_update_write_stamp(cpu_buffer
, event
);
2437 rb_end_commit(cpu_buffer
);
2439 trace_recursive_unlock();
2442 * Only the last preempt count needs to restore preemption.
2444 if (preempt_count() == 1)
2445 ftrace_preempt_enable(per_cpu(rb_need_resched
, cpu
));
2447 preempt_enable_no_resched_notrace();
2450 EXPORT_SYMBOL_GPL(ring_buffer_discard_commit
);
2453 * ring_buffer_write - write data to the buffer without reserving
2454 * @buffer: The ring buffer to write to.
2455 * @length: The length of the data being written (excluding the event header)
2456 * @data: The data to write to the buffer.
2458 * This is like ring_buffer_lock_reserve and ring_buffer_unlock_commit as
2459 * one function. If you already have the data to write to the buffer, it
2460 * may be easier to simply call this function.
2462 * Note, like ring_buffer_lock_reserve, the length is the length of the data
2463 * and not the length of the event which would hold the header.
2465 int ring_buffer_write(struct ring_buffer
*buffer
,
2466 unsigned long length
,
2469 struct ring_buffer_per_cpu
*cpu_buffer
;
2470 struct ring_buffer_event
*event
;
2475 if (ring_buffer_flags
!= RB_BUFFERS_ON
)
2478 if (atomic_read(&buffer
->record_disabled
))
2481 resched
= ftrace_preempt_disable();
2483 cpu
= raw_smp_processor_id();
2485 if (!cpumask_test_cpu(cpu
, buffer
->cpumask
))
2488 cpu_buffer
= buffer
->buffers
[cpu
];
2490 if (atomic_read(&cpu_buffer
->record_disabled
))
2493 if (length
> BUF_MAX_DATA_SIZE
)
2496 event
= rb_reserve_next_event(buffer
, cpu_buffer
, length
);
2500 body
= rb_event_data(event
);
2502 memcpy(body
, data
, length
);
2504 rb_commit(cpu_buffer
, event
);
2508 ftrace_preempt_enable(resched
);
2512 EXPORT_SYMBOL_GPL(ring_buffer_write
);
2514 static int rb_per_cpu_empty(struct ring_buffer_per_cpu
*cpu_buffer
)
2516 struct buffer_page
*reader
= cpu_buffer
->reader_page
;
2517 struct buffer_page
*head
= rb_set_head_page(cpu_buffer
);
2518 struct buffer_page
*commit
= cpu_buffer
->commit_page
;
2520 /* In case of error, head will be NULL */
2521 if (unlikely(!head
))
2524 return reader
->read
== rb_page_commit(reader
) &&
2525 (commit
== reader
||
2527 head
->read
== rb_page_commit(commit
)));
2531 * ring_buffer_record_disable - stop all writes into the buffer
2532 * @buffer: The ring buffer to stop writes to.
2534 * This prevents all writes to the buffer. Any attempt to write
2535 * to the buffer after this will fail and return NULL.
2537 * The caller should call synchronize_sched() after this.
2539 void ring_buffer_record_disable(struct ring_buffer
*buffer
)
2541 atomic_inc(&buffer
->record_disabled
);
2543 EXPORT_SYMBOL_GPL(ring_buffer_record_disable
);
2546 * ring_buffer_record_enable - enable writes to the buffer
2547 * @buffer: The ring buffer to enable writes
2549 * Note, multiple disables will need the same number of enables
2550 * to truely enable the writing (much like preempt_disable).
2552 void ring_buffer_record_enable(struct ring_buffer
*buffer
)
2554 atomic_dec(&buffer
->record_disabled
);
2556 EXPORT_SYMBOL_GPL(ring_buffer_record_enable
);
2559 * ring_buffer_record_disable_cpu - stop all writes into the cpu_buffer
2560 * @buffer: The ring buffer to stop writes to.
2561 * @cpu: The CPU buffer to stop
2563 * This prevents all writes to the buffer. Any attempt to write
2564 * to the buffer after this will fail and return NULL.
2566 * The caller should call synchronize_sched() after this.
2568 void ring_buffer_record_disable_cpu(struct ring_buffer
*buffer
, int cpu
)
2570 struct ring_buffer_per_cpu
*cpu_buffer
;
2572 if (!cpumask_test_cpu(cpu
, buffer
->cpumask
))
2575 cpu_buffer
= buffer
->buffers
[cpu
];
2576 atomic_inc(&cpu_buffer
->record_disabled
);
2578 EXPORT_SYMBOL_GPL(ring_buffer_record_disable_cpu
);
2581 * ring_buffer_record_enable_cpu - enable writes to the buffer
2582 * @buffer: The ring buffer to enable writes
2583 * @cpu: The CPU to enable.
2585 * Note, multiple disables will need the same number of enables
2586 * to truely enable the writing (much like preempt_disable).
2588 void ring_buffer_record_enable_cpu(struct ring_buffer
*buffer
, int cpu
)
2590 struct ring_buffer_per_cpu
*cpu_buffer
;
2592 if (!cpumask_test_cpu(cpu
, buffer
->cpumask
))
2595 cpu_buffer
= buffer
->buffers
[cpu
];
2596 atomic_dec(&cpu_buffer
->record_disabled
);
2598 EXPORT_SYMBOL_GPL(ring_buffer_record_enable_cpu
);
2601 * ring_buffer_entries_cpu - get the number of entries in a cpu buffer
2602 * @buffer: The ring buffer
2603 * @cpu: The per CPU buffer to get the entries from.
2605 unsigned long ring_buffer_entries_cpu(struct ring_buffer
*buffer
, int cpu
)
2607 struct ring_buffer_per_cpu
*cpu_buffer
;
2610 if (!cpumask_test_cpu(cpu
, buffer
->cpumask
))
2613 cpu_buffer
= buffer
->buffers
[cpu
];
2614 ret
= (local_read(&cpu_buffer
->entries
) - local_read(&cpu_buffer
->overrun
))
2619 EXPORT_SYMBOL_GPL(ring_buffer_entries_cpu
);
2622 * ring_buffer_overrun_cpu - get the number of overruns in a cpu_buffer
2623 * @buffer: The ring buffer
2624 * @cpu: The per CPU buffer to get the number of overruns from
2626 unsigned long ring_buffer_overrun_cpu(struct ring_buffer
*buffer
, int cpu
)
2628 struct ring_buffer_per_cpu
*cpu_buffer
;
2631 if (!cpumask_test_cpu(cpu
, buffer
->cpumask
))
2634 cpu_buffer
= buffer
->buffers
[cpu
];
2635 ret
= local_read(&cpu_buffer
->overrun
);
2639 EXPORT_SYMBOL_GPL(ring_buffer_overrun_cpu
);
2642 * ring_buffer_commit_overrun_cpu - get the number of overruns caused by commits
2643 * @buffer: The ring buffer
2644 * @cpu: The per CPU buffer to get the number of overruns from
2647 ring_buffer_commit_overrun_cpu(struct ring_buffer
*buffer
, int cpu
)
2649 struct ring_buffer_per_cpu
*cpu_buffer
;
2652 if (!cpumask_test_cpu(cpu
, buffer
->cpumask
))
2655 cpu_buffer
= buffer
->buffers
[cpu
];
2656 ret
= local_read(&cpu_buffer
->commit_overrun
);
2660 EXPORT_SYMBOL_GPL(ring_buffer_commit_overrun_cpu
);
2663 * ring_buffer_entries - get the number of entries in a buffer
2664 * @buffer: The ring buffer
2666 * Returns the total number of entries in the ring buffer
2669 unsigned long ring_buffer_entries(struct ring_buffer
*buffer
)
2671 struct ring_buffer_per_cpu
*cpu_buffer
;
2672 unsigned long entries
= 0;
2675 /* if you care about this being correct, lock the buffer */
2676 for_each_buffer_cpu(buffer
, cpu
) {
2677 cpu_buffer
= buffer
->buffers
[cpu
];
2678 entries
+= (local_read(&cpu_buffer
->entries
) -
2679 local_read(&cpu_buffer
->overrun
)) - cpu_buffer
->read
;
2684 EXPORT_SYMBOL_GPL(ring_buffer_entries
);
2687 * ring_buffer_overrun_cpu - get the number of overruns in buffer
2688 * @buffer: The ring buffer
2690 * Returns the total number of overruns in the ring buffer
2693 unsigned long ring_buffer_overruns(struct ring_buffer
*buffer
)
2695 struct ring_buffer_per_cpu
*cpu_buffer
;
2696 unsigned long overruns
= 0;
2699 /* if you care about this being correct, lock the buffer */
2700 for_each_buffer_cpu(buffer
, cpu
) {
2701 cpu_buffer
= buffer
->buffers
[cpu
];
2702 overruns
+= local_read(&cpu_buffer
->overrun
);
2707 EXPORT_SYMBOL_GPL(ring_buffer_overruns
);
2709 static void rb_iter_reset(struct ring_buffer_iter
*iter
)
2711 struct ring_buffer_per_cpu
*cpu_buffer
= iter
->cpu_buffer
;
2713 /* Iterator usage is expected to have record disabled */
2714 if (list_empty(&cpu_buffer
->reader_page
->list
)) {
2715 iter
->head_page
= rb_set_head_page(cpu_buffer
);
2716 if (unlikely(!iter
->head_page
))
2718 iter
->head
= iter
->head_page
->read
;
2720 iter
->head_page
= cpu_buffer
->reader_page
;
2721 iter
->head
= cpu_buffer
->reader_page
->read
;
2724 iter
->read_stamp
= cpu_buffer
->read_stamp
;
2726 iter
->read_stamp
= iter
->head_page
->page
->time_stamp
;
2730 * ring_buffer_iter_reset - reset an iterator
2731 * @iter: The iterator to reset
2733 * Resets the iterator, so that it will start from the beginning
2736 void ring_buffer_iter_reset(struct ring_buffer_iter
*iter
)
2738 struct ring_buffer_per_cpu
*cpu_buffer
;
2739 unsigned long flags
;
2744 cpu_buffer
= iter
->cpu_buffer
;
2746 spin_lock_irqsave(&cpu_buffer
->reader_lock
, flags
);
2747 rb_iter_reset(iter
);
2748 spin_unlock_irqrestore(&cpu_buffer
->reader_lock
, flags
);
2750 EXPORT_SYMBOL_GPL(ring_buffer_iter_reset
);
2753 * ring_buffer_iter_empty - check if an iterator has no more to read
2754 * @iter: The iterator to check
2756 int ring_buffer_iter_empty(struct ring_buffer_iter
*iter
)
2758 struct ring_buffer_per_cpu
*cpu_buffer
;
2760 cpu_buffer
= iter
->cpu_buffer
;
2762 return iter
->head_page
== cpu_buffer
->commit_page
&&
2763 iter
->head
== rb_commit_index(cpu_buffer
);
2765 EXPORT_SYMBOL_GPL(ring_buffer_iter_empty
);
2768 rb_update_read_stamp(struct ring_buffer_per_cpu
*cpu_buffer
,
2769 struct ring_buffer_event
*event
)
2773 switch (event
->type_len
) {
2774 case RINGBUF_TYPE_PADDING
:
2777 case RINGBUF_TYPE_TIME_EXTEND
:
2778 delta
= event
->array
[0];
2780 delta
+= event
->time_delta
;
2781 cpu_buffer
->read_stamp
+= delta
;
2784 case RINGBUF_TYPE_TIME_STAMP
:
2785 /* FIXME: not implemented */
2788 case RINGBUF_TYPE_DATA
:
2789 cpu_buffer
->read_stamp
+= event
->time_delta
;
2799 rb_update_iter_read_stamp(struct ring_buffer_iter
*iter
,
2800 struct ring_buffer_event
*event
)
2804 switch (event
->type_len
) {
2805 case RINGBUF_TYPE_PADDING
:
2808 case RINGBUF_TYPE_TIME_EXTEND
:
2809 delta
= event
->array
[0];
2811 delta
+= event
->time_delta
;
2812 iter
->read_stamp
+= delta
;
2815 case RINGBUF_TYPE_TIME_STAMP
:
2816 /* FIXME: not implemented */
2819 case RINGBUF_TYPE_DATA
:
2820 iter
->read_stamp
+= event
->time_delta
;
2829 static struct buffer_page
*
2830 rb_get_reader_page(struct ring_buffer_per_cpu
*cpu_buffer
)
2832 struct buffer_page
*reader
= NULL
;
2833 unsigned long flags
;
2837 local_irq_save(flags
);
2838 __raw_spin_lock(&cpu_buffer
->lock
);
2842 * This should normally only loop twice. But because the
2843 * start of the reader inserts an empty page, it causes
2844 * a case where we will loop three times. There should be no
2845 * reason to loop four times (that I know of).
2847 if (RB_WARN_ON(cpu_buffer
, ++nr_loops
> 3)) {
2852 reader
= cpu_buffer
->reader_page
;
2854 /* If there's more to read, return this page */
2855 if (cpu_buffer
->reader_page
->read
< rb_page_size(reader
))
2858 /* Never should we have an index greater than the size */
2859 if (RB_WARN_ON(cpu_buffer
,
2860 cpu_buffer
->reader_page
->read
> rb_page_size(reader
)))
2863 /* check if we caught up to the tail */
2865 if (cpu_buffer
->commit_page
== cpu_buffer
->reader_page
)
2869 * Reset the reader page to size zero.
2871 local_set(&cpu_buffer
->reader_page
->write
, 0);
2872 local_set(&cpu_buffer
->reader_page
->entries
, 0);
2873 local_set(&cpu_buffer
->reader_page
->page
->commit
, 0);
2877 * Splice the empty reader page into the list around the head.
2879 reader
= rb_set_head_page(cpu_buffer
);
2880 cpu_buffer
->reader_page
->list
.next
= reader
->list
.next
;
2881 cpu_buffer
->reader_page
->list
.prev
= reader
->list
.prev
;
2884 * cpu_buffer->pages just needs to point to the buffer, it
2885 * has no specific buffer page to point to. Lets move it out
2886 * of our way so we don't accidently swap it.
2888 cpu_buffer
->pages
= reader
->list
.prev
;
2890 /* The reader page will be pointing to the new head */
2891 rb_set_list_to_head(cpu_buffer
, &cpu_buffer
->reader_page
->list
);
2894 * Here's the tricky part.
2896 * We need to move the pointer past the header page.
2897 * But we can only do that if a writer is not currently
2898 * moving it. The page before the header page has the
2899 * flag bit '1' set if it is pointing to the page we want.
2900 * but if the writer is in the process of moving it
2901 * than it will be '2' or already moved '0'.
2904 ret
= rb_head_page_replace(reader
, cpu_buffer
->reader_page
);
2907 * If we did not convert it, then we must try again.
2913 * Yeah! We succeeded in replacing the page.
2915 * Now make the new head point back to the reader page.
2917 reader
->list
.next
->prev
= &cpu_buffer
->reader_page
->list
;
2918 rb_inc_page(cpu_buffer
, &cpu_buffer
->head_page
);
2920 /* Finally update the reader page to the new head */
2921 cpu_buffer
->reader_page
= reader
;
2922 rb_reset_reader_page(cpu_buffer
);
2927 __raw_spin_unlock(&cpu_buffer
->lock
);
2928 local_irq_restore(flags
);
2933 static void rb_advance_reader(struct ring_buffer_per_cpu
*cpu_buffer
)
2935 struct ring_buffer_event
*event
;
2936 struct buffer_page
*reader
;
2939 reader
= rb_get_reader_page(cpu_buffer
);
2941 /* This function should not be called when buffer is empty */
2942 if (RB_WARN_ON(cpu_buffer
, !reader
))
2945 event
= rb_reader_event(cpu_buffer
);
2947 if (event
->type_len
<= RINGBUF_TYPE_DATA_TYPE_LEN_MAX
)
2950 rb_update_read_stamp(cpu_buffer
, event
);
2952 length
= rb_event_length(event
);
2953 cpu_buffer
->reader_page
->read
+= length
;
2956 static void rb_advance_iter(struct ring_buffer_iter
*iter
)
2958 struct ring_buffer
*buffer
;
2959 struct ring_buffer_per_cpu
*cpu_buffer
;
2960 struct ring_buffer_event
*event
;
2963 cpu_buffer
= iter
->cpu_buffer
;
2964 buffer
= cpu_buffer
->buffer
;
2967 * Check if we are at the end of the buffer.
2969 if (iter
->head
>= rb_page_size(iter
->head_page
)) {
2970 /* discarded commits can make the page empty */
2971 if (iter
->head_page
== cpu_buffer
->commit_page
)
2977 event
= rb_iter_head_event(iter
);
2979 length
= rb_event_length(event
);
2982 * This should not be called to advance the header if we are
2983 * at the tail of the buffer.
2985 if (RB_WARN_ON(cpu_buffer
,
2986 (iter
->head_page
== cpu_buffer
->commit_page
) &&
2987 (iter
->head
+ length
> rb_commit_index(cpu_buffer
))))
2990 rb_update_iter_read_stamp(iter
, event
);
2992 iter
->head
+= length
;
2994 /* check for end of page padding */
2995 if ((iter
->head
>= rb_page_size(iter
->head_page
)) &&
2996 (iter
->head_page
!= cpu_buffer
->commit_page
))
2997 rb_advance_iter(iter
);
3000 static struct ring_buffer_event
*
3001 rb_buffer_peek(struct ring_buffer_per_cpu
*cpu_buffer
, u64
*ts
)
3003 struct ring_buffer_event
*event
;
3004 struct buffer_page
*reader
;
3009 * We repeat when a timestamp is encountered. It is possible
3010 * to get multiple timestamps from an interrupt entering just
3011 * as one timestamp is about to be written, or from discarded
3012 * commits. The most that we can have is the number on a single page.
3014 if (RB_WARN_ON(cpu_buffer
, ++nr_loops
> RB_TIMESTAMPS_PER_PAGE
))
3017 reader
= rb_get_reader_page(cpu_buffer
);
3021 event
= rb_reader_event(cpu_buffer
);
3023 switch (event
->type_len
) {
3024 case RINGBUF_TYPE_PADDING
:
3025 if (rb_null_event(event
))
3026 RB_WARN_ON(cpu_buffer
, 1);
3028 * Because the writer could be discarding every
3029 * event it creates (which would probably be bad)
3030 * if we were to go back to "again" then we may never
3031 * catch up, and will trigger the warn on, or lock
3032 * the box. Return the padding, and we will release
3033 * the current locks, and try again.
3037 case RINGBUF_TYPE_TIME_EXTEND
:
3038 /* Internal data, OK to advance */
3039 rb_advance_reader(cpu_buffer
);
3042 case RINGBUF_TYPE_TIME_STAMP
:
3043 /* FIXME: not implemented */
3044 rb_advance_reader(cpu_buffer
);
3047 case RINGBUF_TYPE_DATA
:
3049 *ts
= cpu_buffer
->read_stamp
+ event
->time_delta
;
3050 ring_buffer_normalize_time_stamp(cpu_buffer
->buffer
,
3051 cpu_buffer
->cpu
, ts
);
3061 EXPORT_SYMBOL_GPL(ring_buffer_peek
);
3063 static struct ring_buffer_event
*
3064 rb_iter_peek(struct ring_buffer_iter
*iter
, u64
*ts
)
3066 struct ring_buffer
*buffer
;
3067 struct ring_buffer_per_cpu
*cpu_buffer
;
3068 struct ring_buffer_event
*event
;
3071 if (ring_buffer_iter_empty(iter
))
3074 cpu_buffer
= iter
->cpu_buffer
;
3075 buffer
= cpu_buffer
->buffer
;
3079 * We repeat when a timestamp is encountered.
3080 * We can get multiple timestamps by nested interrupts or also
3081 * if filtering is on (discarding commits). Since discarding
3082 * commits can be frequent we can get a lot of timestamps.
3083 * But we limit them by not adding timestamps if they begin
3084 * at the start of a page.
3086 if (RB_WARN_ON(cpu_buffer
, ++nr_loops
> RB_TIMESTAMPS_PER_PAGE
))
3089 if (rb_per_cpu_empty(cpu_buffer
))
3092 event
= rb_iter_head_event(iter
);
3094 switch (event
->type_len
) {
3095 case RINGBUF_TYPE_PADDING
:
3096 if (rb_null_event(event
)) {
3100 rb_advance_iter(iter
);
3103 case RINGBUF_TYPE_TIME_EXTEND
:
3104 /* Internal data, OK to advance */
3105 rb_advance_iter(iter
);
3108 case RINGBUF_TYPE_TIME_STAMP
:
3109 /* FIXME: not implemented */
3110 rb_advance_iter(iter
);
3113 case RINGBUF_TYPE_DATA
:
3115 *ts
= iter
->read_stamp
+ event
->time_delta
;
3116 ring_buffer_normalize_time_stamp(buffer
,
3117 cpu_buffer
->cpu
, ts
);
3127 EXPORT_SYMBOL_GPL(ring_buffer_iter_peek
);
3129 static inline int rb_ok_to_lock(void)
3132 * If an NMI die dumps out the content of the ring buffer
3133 * do not grab locks. We also permanently disable the ring
3134 * buffer too. A one time deal is all you get from reading
3135 * the ring buffer from an NMI.
3137 if (likely(!in_nmi()))
3140 tracing_off_permanent();
3145 * ring_buffer_peek - peek at the next event to be read
3146 * @buffer: The ring buffer to read
3147 * @cpu: The cpu to peak at
3148 * @ts: The timestamp counter of this event.
3150 * This will return the event that will be read next, but does
3151 * not consume the data.
3153 struct ring_buffer_event
*
3154 ring_buffer_peek(struct ring_buffer
*buffer
, int cpu
, u64
*ts
)
3156 struct ring_buffer_per_cpu
*cpu_buffer
= buffer
->buffers
[cpu
];
3157 struct ring_buffer_event
*event
;
3158 unsigned long flags
;
3161 if (!cpumask_test_cpu(cpu
, buffer
->cpumask
))
3164 dolock
= rb_ok_to_lock();
3166 local_irq_save(flags
);
3168 spin_lock(&cpu_buffer
->reader_lock
);
3169 event
= rb_buffer_peek(cpu_buffer
, ts
);
3170 if (event
&& event
->type_len
== RINGBUF_TYPE_PADDING
)
3171 rb_advance_reader(cpu_buffer
);
3173 spin_unlock(&cpu_buffer
->reader_lock
);
3174 local_irq_restore(flags
);
3176 if (event
&& event
->type_len
== RINGBUF_TYPE_PADDING
)
3183 * ring_buffer_iter_peek - peek at the next event to be read
3184 * @iter: The ring buffer iterator
3185 * @ts: The timestamp counter of this event.
3187 * This will return the event that will be read next, but does
3188 * not increment the iterator.
3190 struct ring_buffer_event
*
3191 ring_buffer_iter_peek(struct ring_buffer_iter
*iter
, u64
*ts
)
3193 struct ring_buffer_per_cpu
*cpu_buffer
= iter
->cpu_buffer
;
3194 struct ring_buffer_event
*event
;
3195 unsigned long flags
;
3198 spin_lock_irqsave(&cpu_buffer
->reader_lock
, flags
);
3199 event
= rb_iter_peek(iter
, ts
);
3200 spin_unlock_irqrestore(&cpu_buffer
->reader_lock
, flags
);
3202 if (event
&& event
->type_len
== RINGBUF_TYPE_PADDING
)
3209 * ring_buffer_consume - return an event and consume it
3210 * @buffer: The ring buffer to get the next event from
3212 * Returns the next event in the ring buffer, and that event is consumed.
3213 * Meaning, that sequential reads will keep returning a different event,
3214 * and eventually empty the ring buffer if the producer is slower.
3216 struct ring_buffer_event
*
3217 ring_buffer_consume(struct ring_buffer
*buffer
, int cpu
, u64
*ts
)
3219 struct ring_buffer_per_cpu
*cpu_buffer
;
3220 struct ring_buffer_event
*event
= NULL
;
3221 unsigned long flags
;
3224 dolock
= rb_ok_to_lock();
3227 /* might be called in atomic */
3230 if (!cpumask_test_cpu(cpu
, buffer
->cpumask
))
3233 cpu_buffer
= buffer
->buffers
[cpu
];
3234 local_irq_save(flags
);
3236 spin_lock(&cpu_buffer
->reader_lock
);
3238 event
= rb_buffer_peek(cpu_buffer
, ts
);
3240 rb_advance_reader(cpu_buffer
);
3243 spin_unlock(&cpu_buffer
->reader_lock
);
3244 local_irq_restore(flags
);
3249 if (event
&& event
->type_len
== RINGBUF_TYPE_PADDING
)
3254 EXPORT_SYMBOL_GPL(ring_buffer_consume
);
3257 * ring_buffer_read_start - start a non consuming read of the buffer
3258 * @buffer: The ring buffer to read from
3259 * @cpu: The cpu buffer to iterate over
3261 * This starts up an iteration through the buffer. It also disables
3262 * the recording to the buffer until the reading is finished.
3263 * This prevents the reading from being corrupted. This is not
3264 * a consuming read, so a producer is not expected.
3266 * Must be paired with ring_buffer_finish.
3268 struct ring_buffer_iter
*
3269 ring_buffer_read_start(struct ring_buffer
*buffer
, int cpu
)
3271 struct ring_buffer_per_cpu
*cpu_buffer
;
3272 struct ring_buffer_iter
*iter
;
3273 unsigned long flags
;
3275 if (!cpumask_test_cpu(cpu
, buffer
->cpumask
))
3278 iter
= kmalloc(sizeof(*iter
), GFP_KERNEL
);
3282 cpu_buffer
= buffer
->buffers
[cpu
];
3284 iter
->cpu_buffer
= cpu_buffer
;
3286 atomic_inc(&cpu_buffer
->record_disabled
);
3287 synchronize_sched();
3289 spin_lock_irqsave(&cpu_buffer
->reader_lock
, flags
);
3290 __raw_spin_lock(&cpu_buffer
->lock
);
3291 rb_iter_reset(iter
);
3292 __raw_spin_unlock(&cpu_buffer
->lock
);
3293 spin_unlock_irqrestore(&cpu_buffer
->reader_lock
, flags
);
3297 EXPORT_SYMBOL_GPL(ring_buffer_read_start
);
3300 * ring_buffer_finish - finish reading the iterator of the buffer
3301 * @iter: The iterator retrieved by ring_buffer_start
3303 * This re-enables the recording to the buffer, and frees the
3307 ring_buffer_read_finish(struct ring_buffer_iter
*iter
)
3309 struct ring_buffer_per_cpu
*cpu_buffer
= iter
->cpu_buffer
;
3311 atomic_dec(&cpu_buffer
->record_disabled
);
3314 EXPORT_SYMBOL_GPL(ring_buffer_read_finish
);
3317 * ring_buffer_read - read the next item in the ring buffer by the iterator
3318 * @iter: The ring buffer iterator
3319 * @ts: The time stamp of the event read.
3321 * This reads the next event in the ring buffer and increments the iterator.
3323 struct ring_buffer_event
*
3324 ring_buffer_read(struct ring_buffer_iter
*iter
, u64
*ts
)
3326 struct ring_buffer_event
*event
;
3327 struct ring_buffer_per_cpu
*cpu_buffer
= iter
->cpu_buffer
;
3328 unsigned long flags
;
3330 spin_lock_irqsave(&cpu_buffer
->reader_lock
, flags
);
3332 event
= rb_iter_peek(iter
, ts
);
3336 if (event
->type_len
== RINGBUF_TYPE_PADDING
)
3339 rb_advance_iter(iter
);
3341 spin_unlock_irqrestore(&cpu_buffer
->reader_lock
, flags
);
3345 EXPORT_SYMBOL_GPL(ring_buffer_read
);
3348 * ring_buffer_size - return the size of the ring buffer (in bytes)
3349 * @buffer: The ring buffer.
3351 unsigned long ring_buffer_size(struct ring_buffer
*buffer
)
3353 return BUF_PAGE_SIZE
* buffer
->pages
;
3355 EXPORT_SYMBOL_GPL(ring_buffer_size
);
3358 rb_reset_cpu(struct ring_buffer_per_cpu
*cpu_buffer
)
3360 rb_head_page_deactivate(cpu_buffer
);
3362 cpu_buffer
->head_page
3363 = list_entry(cpu_buffer
->pages
, struct buffer_page
, list
);
3364 local_set(&cpu_buffer
->head_page
->write
, 0);
3365 local_set(&cpu_buffer
->head_page
->entries
, 0);
3366 local_set(&cpu_buffer
->head_page
->page
->commit
, 0);
3368 cpu_buffer
->head_page
->read
= 0;
3370 cpu_buffer
->tail_page
= cpu_buffer
->head_page
;
3371 cpu_buffer
->commit_page
= cpu_buffer
->head_page
;
3373 INIT_LIST_HEAD(&cpu_buffer
->reader_page
->list
);
3374 local_set(&cpu_buffer
->reader_page
->write
, 0);
3375 local_set(&cpu_buffer
->reader_page
->entries
, 0);
3376 local_set(&cpu_buffer
->reader_page
->page
->commit
, 0);
3377 cpu_buffer
->reader_page
->read
= 0;
3379 local_set(&cpu_buffer
->commit_overrun
, 0);
3380 local_set(&cpu_buffer
->overrun
, 0);
3381 local_set(&cpu_buffer
->entries
, 0);
3382 local_set(&cpu_buffer
->committing
, 0);
3383 local_set(&cpu_buffer
->commits
, 0);
3384 cpu_buffer
->read
= 0;
3386 cpu_buffer
->write_stamp
= 0;
3387 cpu_buffer
->read_stamp
= 0;
3389 rb_head_page_activate(cpu_buffer
);
3393 * ring_buffer_reset_cpu - reset a ring buffer per CPU buffer
3394 * @buffer: The ring buffer to reset a per cpu buffer of
3395 * @cpu: The CPU buffer to be reset
3397 void ring_buffer_reset_cpu(struct ring_buffer
*buffer
, int cpu
)
3399 struct ring_buffer_per_cpu
*cpu_buffer
= buffer
->buffers
[cpu
];
3400 unsigned long flags
;
3402 if (!cpumask_test_cpu(cpu
, buffer
->cpumask
))
3405 atomic_inc(&cpu_buffer
->record_disabled
);
3407 spin_lock_irqsave(&cpu_buffer
->reader_lock
, flags
);
3409 if (RB_WARN_ON(cpu_buffer
, local_read(&cpu_buffer
->committing
)))
3412 __raw_spin_lock(&cpu_buffer
->lock
);
3414 rb_reset_cpu(cpu_buffer
);
3416 __raw_spin_unlock(&cpu_buffer
->lock
);
3419 spin_unlock_irqrestore(&cpu_buffer
->reader_lock
, flags
);
3421 atomic_dec(&cpu_buffer
->record_disabled
);
3423 EXPORT_SYMBOL_GPL(ring_buffer_reset_cpu
);
3426 * ring_buffer_reset - reset a ring buffer
3427 * @buffer: The ring buffer to reset all cpu buffers
3429 void ring_buffer_reset(struct ring_buffer
*buffer
)
3433 for_each_buffer_cpu(buffer
, cpu
)
3434 ring_buffer_reset_cpu(buffer
, cpu
);
3436 EXPORT_SYMBOL_GPL(ring_buffer_reset
);
3439 * rind_buffer_empty - is the ring buffer empty?
3440 * @buffer: The ring buffer to test
3442 int ring_buffer_empty(struct ring_buffer
*buffer
)
3444 struct ring_buffer_per_cpu
*cpu_buffer
;
3445 unsigned long flags
;
3450 dolock
= rb_ok_to_lock();
3452 /* yes this is racy, but if you don't like the race, lock the buffer */
3453 for_each_buffer_cpu(buffer
, cpu
) {
3454 cpu_buffer
= buffer
->buffers
[cpu
];
3455 local_irq_save(flags
);
3457 spin_lock(&cpu_buffer
->reader_lock
);
3458 ret
= rb_per_cpu_empty(cpu_buffer
);
3460 spin_unlock(&cpu_buffer
->reader_lock
);
3461 local_irq_restore(flags
);
3469 EXPORT_SYMBOL_GPL(ring_buffer_empty
);
3472 * ring_buffer_empty_cpu - is a cpu buffer of a ring buffer empty?
3473 * @buffer: The ring buffer
3474 * @cpu: The CPU buffer to test
3476 int ring_buffer_empty_cpu(struct ring_buffer
*buffer
, int cpu
)
3478 struct ring_buffer_per_cpu
*cpu_buffer
;
3479 unsigned long flags
;
3483 if (!cpumask_test_cpu(cpu
, buffer
->cpumask
))
3486 dolock
= rb_ok_to_lock();
3488 cpu_buffer
= buffer
->buffers
[cpu
];
3489 local_irq_save(flags
);
3491 spin_lock(&cpu_buffer
->reader_lock
);
3492 ret
= rb_per_cpu_empty(cpu_buffer
);
3494 spin_unlock(&cpu_buffer
->reader_lock
);
3495 local_irq_restore(flags
);
3499 EXPORT_SYMBOL_GPL(ring_buffer_empty_cpu
);
3501 #ifdef CONFIG_RING_BUFFER_ALLOW_SWAP
3503 * ring_buffer_swap_cpu - swap a CPU buffer between two ring buffers
3504 * @buffer_a: One buffer to swap with
3505 * @buffer_b: The other buffer to swap with
3507 * This function is useful for tracers that want to take a "snapshot"
3508 * of a CPU buffer and has another back up buffer lying around.
3509 * it is expected that the tracer handles the cpu buffer not being
3510 * used at the moment.
3512 int ring_buffer_swap_cpu(struct ring_buffer
*buffer_a
,
3513 struct ring_buffer
*buffer_b
, int cpu
)
3515 struct ring_buffer_per_cpu
*cpu_buffer_a
;
3516 struct ring_buffer_per_cpu
*cpu_buffer_b
;
3519 if (!cpumask_test_cpu(cpu
, buffer_a
->cpumask
) ||
3520 !cpumask_test_cpu(cpu
, buffer_b
->cpumask
))
3523 /* At least make sure the two buffers are somewhat the same */
3524 if (buffer_a
->pages
!= buffer_b
->pages
)
3529 if (ring_buffer_flags
!= RB_BUFFERS_ON
)
3532 if (atomic_read(&buffer_a
->record_disabled
))
3535 if (atomic_read(&buffer_b
->record_disabled
))
3538 cpu_buffer_a
= buffer_a
->buffers
[cpu
];
3539 cpu_buffer_b
= buffer_b
->buffers
[cpu
];
3541 if (atomic_read(&cpu_buffer_a
->record_disabled
))
3544 if (atomic_read(&cpu_buffer_b
->record_disabled
))
3548 * We can't do a synchronize_sched here because this
3549 * function can be called in atomic context.
3550 * Normally this will be called from the same CPU as cpu.
3551 * If not it's up to the caller to protect this.
3553 atomic_inc(&cpu_buffer_a
->record_disabled
);
3554 atomic_inc(&cpu_buffer_b
->record_disabled
);
3557 if (local_read(&cpu_buffer_a
->committing
))
3559 if (local_read(&cpu_buffer_b
->committing
))
3562 buffer_a
->buffers
[cpu
] = cpu_buffer_b
;
3563 buffer_b
->buffers
[cpu
] = cpu_buffer_a
;
3565 cpu_buffer_b
->buffer
= buffer_a
;
3566 cpu_buffer_a
->buffer
= buffer_b
;
3571 atomic_dec(&cpu_buffer_a
->record_disabled
);
3572 atomic_dec(&cpu_buffer_b
->record_disabled
);
3576 EXPORT_SYMBOL_GPL(ring_buffer_swap_cpu
);
3577 #endif /* CONFIG_RING_BUFFER_ALLOW_SWAP */
3580 * ring_buffer_alloc_read_page - allocate a page to read from buffer
3581 * @buffer: the buffer to allocate for.
3583 * This function is used in conjunction with ring_buffer_read_page.
3584 * When reading a full page from the ring buffer, these functions
3585 * can be used to speed up the process. The calling function should
3586 * allocate a few pages first with this function. Then when it
3587 * needs to get pages from the ring buffer, it passes the result
3588 * of this function into ring_buffer_read_page, which will swap
3589 * the page that was allocated, with the read page of the buffer.
3592 * The page allocated, or NULL on error.
3594 void *ring_buffer_alloc_read_page(struct ring_buffer
*buffer
)
3596 struct buffer_data_page
*bpage
;
3599 addr
= __get_free_page(GFP_KERNEL
);
3603 bpage
= (void *)addr
;
3605 rb_init_page(bpage
);
3609 EXPORT_SYMBOL_GPL(ring_buffer_alloc_read_page
);
3612 * ring_buffer_free_read_page - free an allocated read page
3613 * @buffer: the buffer the page was allocate for
3614 * @data: the page to free
3616 * Free a page allocated from ring_buffer_alloc_read_page.
3618 void ring_buffer_free_read_page(struct ring_buffer
*buffer
, void *data
)
3620 free_page((unsigned long)data
);
3622 EXPORT_SYMBOL_GPL(ring_buffer_free_read_page
);
3625 * ring_buffer_read_page - extract a page from the ring buffer
3626 * @buffer: buffer to extract from
3627 * @data_page: the page to use allocated from ring_buffer_alloc_read_page
3628 * @len: amount to extract
3629 * @cpu: the cpu of the buffer to extract
3630 * @full: should the extraction only happen when the page is full.
3632 * This function will pull out a page from the ring buffer and consume it.
3633 * @data_page must be the address of the variable that was returned
3634 * from ring_buffer_alloc_read_page. This is because the page might be used
3635 * to swap with a page in the ring buffer.
3638 * rpage = ring_buffer_alloc_read_page(buffer);
3641 * ret = ring_buffer_read_page(buffer, &rpage, len, cpu, 0);
3643 * process_page(rpage, ret);
3645 * When @full is set, the function will not return true unless
3646 * the writer is off the reader page.
3648 * Note: it is up to the calling functions to handle sleeps and wakeups.
3649 * The ring buffer can be used anywhere in the kernel and can not
3650 * blindly call wake_up. The layer that uses the ring buffer must be
3651 * responsible for that.
3654 * >=0 if data has been transferred, returns the offset of consumed data.
3655 * <0 if no data has been transferred.
3657 int ring_buffer_read_page(struct ring_buffer
*buffer
,
3658 void **data_page
, size_t len
, int cpu
, int full
)
3660 struct ring_buffer_per_cpu
*cpu_buffer
= buffer
->buffers
[cpu
];
3661 struct ring_buffer_event
*event
;
3662 struct buffer_data_page
*bpage
;
3663 struct buffer_page
*reader
;
3664 unsigned long flags
;
3665 unsigned int commit
;
3670 if (!cpumask_test_cpu(cpu
, buffer
->cpumask
))
3674 * If len is not big enough to hold the page header, then
3675 * we can not copy anything.
3677 if (len
<= BUF_PAGE_HDR_SIZE
)
3680 len
-= BUF_PAGE_HDR_SIZE
;
3689 spin_lock_irqsave(&cpu_buffer
->reader_lock
, flags
);
3691 reader
= rb_get_reader_page(cpu_buffer
);
3695 event
= rb_reader_event(cpu_buffer
);
3697 read
= reader
->read
;
3698 commit
= rb_page_commit(reader
);
3701 * If this page has been partially read or
3702 * if len is not big enough to read the rest of the page or
3703 * a writer is still on the page, then
3704 * we must copy the data from the page to the buffer.
3705 * Otherwise, we can simply swap the page with the one passed in.
3707 if (read
|| (len
< (commit
- read
)) ||
3708 cpu_buffer
->reader_page
== cpu_buffer
->commit_page
) {
3709 struct buffer_data_page
*rpage
= cpu_buffer
->reader_page
->page
;
3710 unsigned int rpos
= read
;
3711 unsigned int pos
= 0;
3717 if (len
> (commit
- read
))
3718 len
= (commit
- read
);
3720 size
= rb_event_length(event
);
3725 /* save the current timestamp, since the user will need it */
3726 save_timestamp
= cpu_buffer
->read_stamp
;
3728 /* Need to copy one event at a time */
3730 memcpy(bpage
->data
+ pos
, rpage
->data
+ rpos
, size
);
3734 rb_advance_reader(cpu_buffer
);
3735 rpos
= reader
->read
;
3738 event
= rb_reader_event(cpu_buffer
);
3739 size
= rb_event_length(event
);
3740 } while (len
> size
);
3743 local_set(&bpage
->commit
, pos
);
3744 bpage
->time_stamp
= save_timestamp
;
3746 /* we copied everything to the beginning */
3749 /* update the entry counter */
3750 cpu_buffer
->read
+= rb_page_entries(reader
);
3752 /* swap the pages */
3753 rb_init_page(bpage
);
3754 bpage
= reader
->page
;
3755 reader
->page
= *data_page
;
3756 local_set(&reader
->write
, 0);
3757 local_set(&reader
->entries
, 0);
3764 spin_unlock_irqrestore(&cpu_buffer
->reader_lock
, flags
);
3769 EXPORT_SYMBOL_GPL(ring_buffer_read_page
);
3771 #ifdef CONFIG_TRACING
3773 rb_simple_read(struct file
*filp
, char __user
*ubuf
,
3774 size_t cnt
, loff_t
*ppos
)
3776 unsigned long *p
= filp
->private_data
;
3780 if (test_bit(RB_BUFFERS_DISABLED_BIT
, p
))
3781 r
= sprintf(buf
, "permanently disabled\n");
3783 r
= sprintf(buf
, "%d\n", test_bit(RB_BUFFERS_ON_BIT
, p
));
3785 return simple_read_from_buffer(ubuf
, cnt
, ppos
, buf
, r
);
3789 rb_simple_write(struct file
*filp
, const char __user
*ubuf
,
3790 size_t cnt
, loff_t
*ppos
)
3792 unsigned long *p
= filp
->private_data
;
3797 if (cnt
>= sizeof(buf
))
3800 if (copy_from_user(&buf
, ubuf
, cnt
))
3805 ret
= strict_strtoul(buf
, 10, &val
);
3810 set_bit(RB_BUFFERS_ON_BIT
, p
);
3812 clear_bit(RB_BUFFERS_ON_BIT
, p
);
3819 static const struct file_operations rb_simple_fops
= {
3820 .open
= tracing_open_generic
,
3821 .read
= rb_simple_read
,
3822 .write
= rb_simple_write
,
3826 static __init
int rb_init_debugfs(void)
3828 struct dentry
*d_tracer
;
3830 d_tracer
= tracing_init_dentry();
3832 trace_create_file("tracing_on", 0644, d_tracer
,
3833 &ring_buffer_flags
, &rb_simple_fops
);
3838 fs_initcall(rb_init_debugfs
);
3841 #ifdef CONFIG_HOTPLUG_CPU
3842 static int rb_cpu_notify(struct notifier_block
*self
,
3843 unsigned long action
, void *hcpu
)
3845 struct ring_buffer
*buffer
=
3846 container_of(self
, struct ring_buffer
, cpu_notify
);
3847 long cpu
= (long)hcpu
;
3850 case CPU_UP_PREPARE
:
3851 case CPU_UP_PREPARE_FROZEN
:
3852 if (cpumask_test_cpu(cpu
, buffer
->cpumask
))
3855 buffer
->buffers
[cpu
] =
3856 rb_allocate_cpu_buffer(buffer
, cpu
);
3857 if (!buffer
->buffers
[cpu
]) {
3858 WARN(1, "failed to allocate ring buffer on CPU %ld\n",
3863 cpumask_set_cpu(cpu
, buffer
->cpumask
);
3865 case CPU_DOWN_PREPARE
:
3866 case CPU_DOWN_PREPARE_FROZEN
:
3869 * If we were to free the buffer, then the user would
3870 * lose any trace that was in the buffer.