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