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 */ 30 int 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 146 enum { 147 RB_BUFFERS_ON_BIT = 0, 148 RB_BUFFERS_DISABLED_BIT = 1, 149 }; 150 151 enum { 152 RB_BUFFERS_ON = 1 << RB_BUFFERS_ON_BIT, 153 RB_BUFFERS_DISABLED = 1 << RB_BUFFERS_DISABLED_BIT, 154 }; 155 156 static 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 */ 166 void tracing_on(void) 167 { 168 set_bit(RB_BUFFERS_ON_BIT, &ring_buffer_flags); 169 } 170 EXPORT_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 */ 180 void tracing_off(void) 181 { 182 clear_bit(RB_BUFFERS_ON_BIT, &ring_buffer_flags); 183 } 184 EXPORT_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 */ 192 void 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 */ 200 int tracing_is_on(void) 201 { 202 return ring_buffer_flags == RB_BUFFERS_ON; 203 } 204 EXPORT_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 222 enum { 223 RB_LEN_TIME_EXTEND = 8, 224 RB_LEN_TIME_STAMP = 16, 225 }; 226 227 static inline int rb_null_event(struct ring_buffer_event *event) 228 { 229 return event->type_len == RINGBUF_TYPE_PADDING && !event->time_delta; 230 } 231 232 static 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 239 static unsigned 240 rb_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 */ 252 static unsigned 253 rb_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 */ 281 unsigned 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 } 291 EXPORT_SYMBOL_GPL(ring_buffer_event_length); 292 293 /* inline for ring buffer fast paths */ 294 static void * 295 rb_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 */ 309 void *ring_buffer_event_data(struct ring_buffer_event *event) 310 { 311 return rb_event_data(event); 312 } 313 EXPORT_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 327 struct 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 */ 341 struct 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 365 static 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 */ 376 size_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 */ 386 static 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 */ 395 static 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_STAMP) 409 410 int 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 */ 444 struct ring_buffer_per_cpu { 445 int cpu; 446 struct ring_buffer *buffer; 447 spinlock_t reader_lock; /* serialize readers */ 448 arch_spinlock_t lock; 449 struct lock_class_key lock_key; 450 struct list_head *pages; 451 struct buffer_page *head_page; /* read from head */ 452 struct buffer_page *tail_page; /* write to tail */ 453 struct buffer_page *commit_page; /* committed pages */ 454 struct buffer_page *reader_page; 455 unsigned long lost_events; 456 unsigned long last_overrun; 457 local_t commit_overrun; 458 local_t overrun; 459 local_t entries; 460 local_t committing; 461 local_t commits; 462 unsigned long read; 463 u64 write_stamp; 464 u64 read_stamp; 465 atomic_t record_disabled; 466 }; 467 468 struct 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 487 struct 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 515 static 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 521 u64 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 } 531 EXPORT_SYMBOL_GPL(ring_buffer_time_stamp); 532 533 void 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 } 539 EXPORT_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 */ 623 static 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 */ 638 static int inline 639 rb_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 */ 659 static 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 */ 669 static 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 */ 682 static 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 696 static 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 */ 706 static void 707 rb_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 718 static 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 741 static 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 750 static 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 759 static 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 768 static 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 776 static struct buffer_page * 777 rb_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 814 static 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 */ 834 static 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 901 static 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 */ 915 static 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 */ 932 static 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 963 static 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 1010 static struct ring_buffer_per_cpu * 1011 rb_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 1065 static 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 1087 static 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 */ 1101 struct 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 } 1178 EXPORT_SYMBOL_GPL(__ring_buffer_alloc); 1179 1180 /** 1181 * ring_buffer_free - free a ring buffer. 1182 * @buffer: the buffer to free. 1183 */ 1184 void 1185 ring_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 } 1205 EXPORT_SYMBOL_GPL(ring_buffer_free); 1206 1207 void ring_buffer_set_clock(struct ring_buffer *buffer, 1208 u64 (*clock)(void)) 1209 { 1210 buffer->clock = clock; 1211 } 1212 1213 static void rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer); 1214 1215 static void 1216 rb_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 1239 out: 1240 spin_unlock_irq(&cpu_buffer->reader_lock); 1241 } 1242 1243 static void 1244 rb_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 1265 out: 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 */ 1278 int 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 } 1396 EXPORT_SYMBOL_GPL(ring_buffer_resize); 1397 1398 static inline void * 1399 __rb_data_page_index(struct buffer_data_page *bpage, unsigned index) 1400 { 1401 return bpage->data + index; 1402 } 1403 1404 static inline void *__rb_page_index(struct buffer_page *bpage, unsigned index) 1405 { 1406 return bpage->page->data + index; 1407 } 1408 1409 static inline struct ring_buffer_event * 1410 rb_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 1416 static inline struct ring_buffer_event * 1417 rb_iter_head_event(struct ring_buffer_iter *iter) 1418 { 1419 return __rb_page_index(iter->head_page, iter->head); 1420 } 1421 1422 static inline unsigned long rb_page_write(struct buffer_page *bpage) 1423 { 1424 return local_read(&bpage->write) & RB_WRITE_MASK; 1425 } 1426 1427 static inline unsigned rb_page_commit(struct buffer_page *bpage) 1428 { 1429 return local_read(&bpage->page->commit); 1430 } 1431 1432 static 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 */ 1438 static inline unsigned rb_page_size(struct buffer_page *bpage) 1439 { 1440 return rb_page_commit(bpage); 1441 } 1442 1443 static inline unsigned 1444 rb_commit_index(struct ring_buffer_per_cpu *cpu_buffer) 1445 { 1446 return rb_page_commit(cpu_buffer->commit_page); 1447 } 1448 1449 static inline unsigned 1450 rb_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 1457 static inline int 1458 rb_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 1471 static void 1472 rb_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 1524 static 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 1530 static 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 */ 1560 static void 1561 rb_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 */ 1592 static int 1593 rb_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 1742 static 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 1759 static inline void 1760 rb_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 1826 static struct ring_buffer_event * 1827 rb_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 1930 static 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 1970 static inline int 1971 rb_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 2006 static int 2007 rb_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 2069 static 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 2075 static 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 2107 static struct ring_buffer_event * 2108 rb_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 2210 static 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 2231 static 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 static DEFINE_PER_CPU(int, rb_need_resched); 2246 2247 /** 2248 * ring_buffer_lock_reserve - reserve a part of the buffer 2249 * @buffer: the ring buffer to reserve from 2250 * @length: the length of the data to reserve (excluding event header) 2251 * 2252 * Returns a reseverd event on the ring buffer to copy directly to. 2253 * The user of this interface will need to get the body to write into 2254 * and can use the ring_buffer_event_data() interface. 2255 * 2256 * The length is the length of the data needed, not the event length 2257 * which also includes the event header. 2258 * 2259 * Must be paired with ring_buffer_unlock_commit, unless NULL is returned. 2260 * If NULL is returned, then nothing has been allocated or locked. 2261 */ 2262 struct ring_buffer_event * 2263 ring_buffer_lock_reserve(struct ring_buffer *buffer, unsigned long length) 2264 { 2265 struct ring_buffer_per_cpu *cpu_buffer; 2266 struct ring_buffer_event *event; 2267 int cpu, resched; 2268 2269 if (ring_buffer_flags != RB_BUFFERS_ON) 2270 return NULL; 2271 2272 /* If we are tracing schedule, we don't want to recurse */ 2273 resched = ftrace_preempt_disable(); 2274 2275 if (atomic_read(&buffer->record_disabled)) 2276 goto out_nocheck; 2277 2278 if (trace_recursive_lock()) 2279 goto out_nocheck; 2280 2281 cpu = raw_smp_processor_id(); 2282 2283 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 2284 goto out; 2285 2286 cpu_buffer = buffer->buffers[cpu]; 2287 2288 if (atomic_read(&cpu_buffer->record_disabled)) 2289 goto out; 2290 2291 if (length > BUF_MAX_DATA_SIZE) 2292 goto out; 2293 2294 event = rb_reserve_next_event(buffer, cpu_buffer, length); 2295 if (!event) 2296 goto out; 2297 2298 /* 2299 * Need to store resched state on this cpu. 2300 * Only the first needs to. 2301 */ 2302 2303 if (preempt_count() == 1) 2304 per_cpu(rb_need_resched, cpu) = resched; 2305 2306 return event; 2307 2308 out: 2309 trace_recursive_unlock(); 2310 2311 out_nocheck: 2312 ftrace_preempt_enable(resched); 2313 return NULL; 2314 } 2315 EXPORT_SYMBOL_GPL(ring_buffer_lock_reserve); 2316 2317 static void 2318 rb_update_write_stamp(struct ring_buffer_per_cpu *cpu_buffer, 2319 struct ring_buffer_event *event) 2320 { 2321 /* 2322 * The event first in the commit queue updates the 2323 * time stamp. 2324 */ 2325 if (rb_event_is_commit(cpu_buffer, event)) 2326 cpu_buffer->write_stamp += event->time_delta; 2327 } 2328 2329 static void rb_commit(struct ring_buffer_per_cpu *cpu_buffer, 2330 struct ring_buffer_event *event) 2331 { 2332 local_inc(&cpu_buffer->entries); 2333 rb_update_write_stamp(cpu_buffer, event); 2334 rb_end_commit(cpu_buffer); 2335 } 2336 2337 /** 2338 * ring_buffer_unlock_commit - commit a reserved 2339 * @buffer: The buffer to commit to 2340 * @event: The event pointer to commit. 2341 * 2342 * This commits the data to the ring buffer, and releases any locks held. 2343 * 2344 * Must be paired with ring_buffer_lock_reserve. 2345 */ 2346 int ring_buffer_unlock_commit(struct ring_buffer *buffer, 2347 struct ring_buffer_event *event) 2348 { 2349 struct ring_buffer_per_cpu *cpu_buffer; 2350 int cpu = raw_smp_processor_id(); 2351 2352 cpu_buffer = buffer->buffers[cpu]; 2353 2354 rb_commit(cpu_buffer, event); 2355 2356 trace_recursive_unlock(); 2357 2358 /* 2359 * Only the last preempt count needs to restore preemption. 2360 */ 2361 if (preempt_count() == 1) 2362 ftrace_preempt_enable(per_cpu(rb_need_resched, cpu)); 2363 else 2364 preempt_enable_no_resched_notrace(); 2365 2366 return 0; 2367 } 2368 EXPORT_SYMBOL_GPL(ring_buffer_unlock_commit); 2369 2370 static inline void rb_event_discard(struct ring_buffer_event *event) 2371 { 2372 /* array[0] holds the actual length for the discarded event */ 2373 event->array[0] = rb_event_data_length(event) - RB_EVNT_HDR_SIZE; 2374 event->type_len = RINGBUF_TYPE_PADDING; 2375 /* time delta must be non zero */ 2376 if (!event->time_delta) 2377 event->time_delta = 1; 2378 } 2379 2380 /* 2381 * Decrement the entries to the page that an event is on. 2382 * The event does not even need to exist, only the pointer 2383 * to the page it is on. This may only be called before the commit 2384 * takes place. 2385 */ 2386 static inline void 2387 rb_decrement_entry(struct ring_buffer_per_cpu *cpu_buffer, 2388 struct ring_buffer_event *event) 2389 { 2390 unsigned long addr = (unsigned long)event; 2391 struct buffer_page *bpage = cpu_buffer->commit_page; 2392 struct buffer_page *start; 2393 2394 addr &= PAGE_MASK; 2395 2396 /* Do the likely case first */ 2397 if (likely(bpage->page == (void *)addr)) { 2398 local_dec(&bpage->entries); 2399 return; 2400 } 2401 2402 /* 2403 * Because the commit page may be on the reader page we 2404 * start with the next page and check the end loop there. 2405 */ 2406 rb_inc_page(cpu_buffer, &bpage); 2407 start = bpage; 2408 do { 2409 if (bpage->page == (void *)addr) { 2410 local_dec(&bpage->entries); 2411 return; 2412 } 2413 rb_inc_page(cpu_buffer, &bpage); 2414 } while (bpage != start); 2415 2416 /* commit not part of this buffer?? */ 2417 RB_WARN_ON(cpu_buffer, 1); 2418 } 2419 2420 /** 2421 * ring_buffer_commit_discard - discard an event that has not been committed 2422 * @buffer: the ring buffer 2423 * @event: non committed event to discard 2424 * 2425 * Sometimes an event that is in the ring buffer needs to be ignored. 2426 * This function lets the user discard an event in the ring buffer 2427 * and then that event will not be read later. 2428 * 2429 * This function only works if it is called before the the item has been 2430 * committed. It will try to free the event from the ring buffer 2431 * if another event has not been added behind it. 2432 * 2433 * If another event has been added behind it, it will set the event 2434 * up as discarded, and perform the commit. 2435 * 2436 * If this function is called, do not call ring_buffer_unlock_commit on 2437 * the event. 2438 */ 2439 void ring_buffer_discard_commit(struct ring_buffer *buffer, 2440 struct ring_buffer_event *event) 2441 { 2442 struct ring_buffer_per_cpu *cpu_buffer; 2443 int cpu; 2444 2445 /* The event is discarded regardless */ 2446 rb_event_discard(event); 2447 2448 cpu = smp_processor_id(); 2449 cpu_buffer = buffer->buffers[cpu]; 2450 2451 /* 2452 * This must only be called if the event has not been 2453 * committed yet. Thus we can assume that preemption 2454 * is still disabled. 2455 */ 2456 RB_WARN_ON(buffer, !local_read(&cpu_buffer->committing)); 2457 2458 rb_decrement_entry(cpu_buffer, event); 2459 if (rb_try_to_discard(cpu_buffer, event)) 2460 goto out; 2461 2462 /* 2463 * The commit is still visible by the reader, so we 2464 * must still update the timestamp. 2465 */ 2466 rb_update_write_stamp(cpu_buffer, event); 2467 out: 2468 rb_end_commit(cpu_buffer); 2469 2470 trace_recursive_unlock(); 2471 2472 /* 2473 * Only the last preempt count needs to restore preemption. 2474 */ 2475 if (preempt_count() == 1) 2476 ftrace_preempt_enable(per_cpu(rb_need_resched, cpu)); 2477 else 2478 preempt_enable_no_resched_notrace(); 2479 2480 } 2481 EXPORT_SYMBOL_GPL(ring_buffer_discard_commit); 2482 2483 /** 2484 * ring_buffer_write - write data to the buffer without reserving 2485 * @buffer: The ring buffer to write to. 2486 * @length: The length of the data being written (excluding the event header) 2487 * @data: The data to write to the buffer. 2488 * 2489 * This is like ring_buffer_lock_reserve and ring_buffer_unlock_commit as 2490 * one function. If you already have the data to write to the buffer, it 2491 * may be easier to simply call this function. 2492 * 2493 * Note, like ring_buffer_lock_reserve, the length is the length of the data 2494 * and not the length of the event which would hold the header. 2495 */ 2496 int ring_buffer_write(struct ring_buffer *buffer, 2497 unsigned long length, 2498 void *data) 2499 { 2500 struct ring_buffer_per_cpu *cpu_buffer; 2501 struct ring_buffer_event *event; 2502 void *body; 2503 int ret = -EBUSY; 2504 int cpu, resched; 2505 2506 if (ring_buffer_flags != RB_BUFFERS_ON) 2507 return -EBUSY; 2508 2509 resched = ftrace_preempt_disable(); 2510 2511 if (atomic_read(&buffer->record_disabled)) 2512 goto out; 2513 2514 cpu = raw_smp_processor_id(); 2515 2516 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 2517 goto out; 2518 2519 cpu_buffer = buffer->buffers[cpu]; 2520 2521 if (atomic_read(&cpu_buffer->record_disabled)) 2522 goto out; 2523 2524 if (length > BUF_MAX_DATA_SIZE) 2525 goto out; 2526 2527 event = rb_reserve_next_event(buffer, cpu_buffer, length); 2528 if (!event) 2529 goto out; 2530 2531 body = rb_event_data(event); 2532 2533 memcpy(body, data, length); 2534 2535 rb_commit(cpu_buffer, event); 2536 2537 ret = 0; 2538 out: 2539 ftrace_preempt_enable(resched); 2540 2541 return ret; 2542 } 2543 EXPORT_SYMBOL_GPL(ring_buffer_write); 2544 2545 static int rb_per_cpu_empty(struct ring_buffer_per_cpu *cpu_buffer) 2546 { 2547 struct buffer_page *reader = cpu_buffer->reader_page; 2548 struct buffer_page *head = rb_set_head_page(cpu_buffer); 2549 struct buffer_page *commit = cpu_buffer->commit_page; 2550 2551 /* In case of error, head will be NULL */ 2552 if (unlikely(!head)) 2553 return 1; 2554 2555 return reader->read == rb_page_commit(reader) && 2556 (commit == reader || 2557 (commit == head && 2558 head->read == rb_page_commit(commit))); 2559 } 2560 2561 /** 2562 * ring_buffer_record_disable - stop all writes into the buffer 2563 * @buffer: The ring buffer to stop writes to. 2564 * 2565 * This prevents all writes to the buffer. Any attempt to write 2566 * to the buffer after this will fail and return NULL. 2567 * 2568 * The caller should call synchronize_sched() after this. 2569 */ 2570 void ring_buffer_record_disable(struct ring_buffer *buffer) 2571 { 2572 atomic_inc(&buffer->record_disabled); 2573 } 2574 EXPORT_SYMBOL_GPL(ring_buffer_record_disable); 2575 2576 /** 2577 * ring_buffer_record_enable - enable writes to the buffer 2578 * @buffer: The ring buffer to enable writes 2579 * 2580 * Note, multiple disables will need the same number of enables 2581 * to truly enable the writing (much like preempt_disable). 2582 */ 2583 void ring_buffer_record_enable(struct ring_buffer *buffer) 2584 { 2585 atomic_dec(&buffer->record_disabled); 2586 } 2587 EXPORT_SYMBOL_GPL(ring_buffer_record_enable); 2588 2589 /** 2590 * ring_buffer_record_disable_cpu - stop all writes into the cpu_buffer 2591 * @buffer: The ring buffer to stop writes to. 2592 * @cpu: The CPU buffer to stop 2593 * 2594 * This prevents all writes to the buffer. Any attempt to write 2595 * to the buffer after this will fail and return NULL. 2596 * 2597 * The caller should call synchronize_sched() after this. 2598 */ 2599 void ring_buffer_record_disable_cpu(struct ring_buffer *buffer, int cpu) 2600 { 2601 struct ring_buffer_per_cpu *cpu_buffer; 2602 2603 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 2604 return; 2605 2606 cpu_buffer = buffer->buffers[cpu]; 2607 atomic_inc(&cpu_buffer->record_disabled); 2608 } 2609 EXPORT_SYMBOL_GPL(ring_buffer_record_disable_cpu); 2610 2611 /** 2612 * ring_buffer_record_enable_cpu - enable writes to the buffer 2613 * @buffer: The ring buffer to enable writes 2614 * @cpu: The CPU to enable. 2615 * 2616 * Note, multiple disables will need the same number of enables 2617 * to truly enable the writing (much like preempt_disable). 2618 */ 2619 void ring_buffer_record_enable_cpu(struct ring_buffer *buffer, int cpu) 2620 { 2621 struct ring_buffer_per_cpu *cpu_buffer; 2622 2623 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 2624 return; 2625 2626 cpu_buffer = buffer->buffers[cpu]; 2627 atomic_dec(&cpu_buffer->record_disabled); 2628 } 2629 EXPORT_SYMBOL_GPL(ring_buffer_record_enable_cpu); 2630 2631 /** 2632 * ring_buffer_entries_cpu - get the number of entries in a cpu buffer 2633 * @buffer: The ring buffer 2634 * @cpu: The per CPU buffer to get the entries from. 2635 */ 2636 unsigned long ring_buffer_entries_cpu(struct ring_buffer *buffer, int cpu) 2637 { 2638 struct ring_buffer_per_cpu *cpu_buffer; 2639 unsigned long ret; 2640 2641 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 2642 return 0; 2643 2644 cpu_buffer = buffer->buffers[cpu]; 2645 ret = (local_read(&cpu_buffer->entries) - local_read(&cpu_buffer->overrun)) 2646 - cpu_buffer->read; 2647 2648 return ret; 2649 } 2650 EXPORT_SYMBOL_GPL(ring_buffer_entries_cpu); 2651 2652 /** 2653 * ring_buffer_overrun_cpu - get the number of overruns in a cpu_buffer 2654 * @buffer: The ring buffer 2655 * @cpu: The per CPU buffer to get the number of overruns from 2656 */ 2657 unsigned long ring_buffer_overrun_cpu(struct ring_buffer *buffer, int cpu) 2658 { 2659 struct ring_buffer_per_cpu *cpu_buffer; 2660 unsigned long ret; 2661 2662 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 2663 return 0; 2664 2665 cpu_buffer = buffer->buffers[cpu]; 2666 ret = local_read(&cpu_buffer->overrun); 2667 2668 return ret; 2669 } 2670 EXPORT_SYMBOL_GPL(ring_buffer_overrun_cpu); 2671 2672 /** 2673 * ring_buffer_commit_overrun_cpu - get the number of overruns caused by commits 2674 * @buffer: The ring buffer 2675 * @cpu: The per CPU buffer to get the number of overruns from 2676 */ 2677 unsigned long 2678 ring_buffer_commit_overrun_cpu(struct ring_buffer *buffer, int cpu) 2679 { 2680 struct ring_buffer_per_cpu *cpu_buffer; 2681 unsigned long ret; 2682 2683 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 2684 return 0; 2685 2686 cpu_buffer = buffer->buffers[cpu]; 2687 ret = local_read(&cpu_buffer->commit_overrun); 2688 2689 return ret; 2690 } 2691 EXPORT_SYMBOL_GPL(ring_buffer_commit_overrun_cpu); 2692 2693 /** 2694 * ring_buffer_entries - get the number of entries in a buffer 2695 * @buffer: The ring buffer 2696 * 2697 * Returns the total number of entries in the ring buffer 2698 * (all CPU entries) 2699 */ 2700 unsigned long ring_buffer_entries(struct ring_buffer *buffer) 2701 { 2702 struct ring_buffer_per_cpu *cpu_buffer; 2703 unsigned long entries = 0; 2704 int cpu; 2705 2706 /* if you care about this being correct, lock the buffer */ 2707 for_each_buffer_cpu(buffer, cpu) { 2708 cpu_buffer = buffer->buffers[cpu]; 2709 entries += (local_read(&cpu_buffer->entries) - 2710 local_read(&cpu_buffer->overrun)) - cpu_buffer->read; 2711 } 2712 2713 return entries; 2714 } 2715 EXPORT_SYMBOL_GPL(ring_buffer_entries); 2716 2717 /** 2718 * ring_buffer_overruns - get the number of overruns in buffer 2719 * @buffer: The ring buffer 2720 * 2721 * Returns the total number of overruns in the ring buffer 2722 * (all CPU entries) 2723 */ 2724 unsigned long ring_buffer_overruns(struct ring_buffer *buffer) 2725 { 2726 struct ring_buffer_per_cpu *cpu_buffer; 2727 unsigned long overruns = 0; 2728 int cpu; 2729 2730 /* if you care about this being correct, lock the buffer */ 2731 for_each_buffer_cpu(buffer, cpu) { 2732 cpu_buffer = buffer->buffers[cpu]; 2733 overruns += local_read(&cpu_buffer->overrun); 2734 } 2735 2736 return overruns; 2737 } 2738 EXPORT_SYMBOL_GPL(ring_buffer_overruns); 2739 2740 static void rb_iter_reset(struct ring_buffer_iter *iter) 2741 { 2742 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer; 2743 2744 /* Iterator usage is expected to have record disabled */ 2745 if (list_empty(&cpu_buffer->reader_page->list)) { 2746 iter->head_page = rb_set_head_page(cpu_buffer); 2747 if (unlikely(!iter->head_page)) 2748 return; 2749 iter->head = iter->head_page->read; 2750 } else { 2751 iter->head_page = cpu_buffer->reader_page; 2752 iter->head = cpu_buffer->reader_page->read; 2753 } 2754 if (iter->head) 2755 iter->read_stamp = cpu_buffer->read_stamp; 2756 else 2757 iter->read_stamp = iter->head_page->page->time_stamp; 2758 iter->cache_reader_page = cpu_buffer->reader_page; 2759 iter->cache_read = cpu_buffer->read; 2760 } 2761 2762 /** 2763 * ring_buffer_iter_reset - reset an iterator 2764 * @iter: The iterator to reset 2765 * 2766 * Resets the iterator, so that it will start from the beginning 2767 * again. 2768 */ 2769 void ring_buffer_iter_reset(struct ring_buffer_iter *iter) 2770 { 2771 struct ring_buffer_per_cpu *cpu_buffer; 2772 unsigned long flags; 2773 2774 if (!iter) 2775 return; 2776 2777 cpu_buffer = iter->cpu_buffer; 2778 2779 spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 2780 rb_iter_reset(iter); 2781 spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 2782 } 2783 EXPORT_SYMBOL_GPL(ring_buffer_iter_reset); 2784 2785 /** 2786 * ring_buffer_iter_empty - check if an iterator has no more to read 2787 * @iter: The iterator to check 2788 */ 2789 int ring_buffer_iter_empty(struct ring_buffer_iter *iter) 2790 { 2791 struct ring_buffer_per_cpu *cpu_buffer; 2792 2793 cpu_buffer = iter->cpu_buffer; 2794 2795 return iter->head_page == cpu_buffer->commit_page && 2796 iter->head == rb_commit_index(cpu_buffer); 2797 } 2798 EXPORT_SYMBOL_GPL(ring_buffer_iter_empty); 2799 2800 static void 2801 rb_update_read_stamp(struct ring_buffer_per_cpu *cpu_buffer, 2802 struct ring_buffer_event *event) 2803 { 2804 u64 delta; 2805 2806 switch (event->type_len) { 2807 case RINGBUF_TYPE_PADDING: 2808 return; 2809 2810 case RINGBUF_TYPE_TIME_EXTEND: 2811 delta = event->array[0]; 2812 delta <<= TS_SHIFT; 2813 delta += event->time_delta; 2814 cpu_buffer->read_stamp += delta; 2815 return; 2816 2817 case RINGBUF_TYPE_TIME_STAMP: 2818 /* FIXME: not implemented */ 2819 return; 2820 2821 case RINGBUF_TYPE_DATA: 2822 cpu_buffer->read_stamp += event->time_delta; 2823 return; 2824 2825 default: 2826 BUG(); 2827 } 2828 return; 2829 } 2830 2831 static void 2832 rb_update_iter_read_stamp(struct ring_buffer_iter *iter, 2833 struct ring_buffer_event *event) 2834 { 2835 u64 delta; 2836 2837 switch (event->type_len) { 2838 case RINGBUF_TYPE_PADDING: 2839 return; 2840 2841 case RINGBUF_TYPE_TIME_EXTEND: 2842 delta = event->array[0]; 2843 delta <<= TS_SHIFT; 2844 delta += event->time_delta; 2845 iter->read_stamp += delta; 2846 return; 2847 2848 case RINGBUF_TYPE_TIME_STAMP: 2849 /* FIXME: not implemented */ 2850 return; 2851 2852 case RINGBUF_TYPE_DATA: 2853 iter->read_stamp += event->time_delta; 2854 return; 2855 2856 default: 2857 BUG(); 2858 } 2859 return; 2860 } 2861 2862 static struct buffer_page * 2863 rb_get_reader_page(struct ring_buffer_per_cpu *cpu_buffer) 2864 { 2865 struct buffer_page *reader = NULL; 2866 unsigned long overwrite; 2867 unsigned long flags; 2868 int nr_loops = 0; 2869 int ret; 2870 2871 local_irq_save(flags); 2872 arch_spin_lock(&cpu_buffer->lock); 2873 2874 again: 2875 /* 2876 * This should normally only loop twice. But because the 2877 * start of the reader inserts an empty page, it causes 2878 * a case where we will loop three times. There should be no 2879 * reason to loop four times (that I know of). 2880 */ 2881 if (RB_WARN_ON(cpu_buffer, ++nr_loops > 3)) { 2882 reader = NULL; 2883 goto out; 2884 } 2885 2886 reader = cpu_buffer->reader_page; 2887 2888 /* If there's more to read, return this page */ 2889 if (cpu_buffer->reader_page->read < rb_page_size(reader)) 2890 goto out; 2891 2892 /* Never should we have an index greater than the size */ 2893 if (RB_WARN_ON(cpu_buffer, 2894 cpu_buffer->reader_page->read > rb_page_size(reader))) 2895 goto out; 2896 2897 /* check if we caught up to the tail */ 2898 reader = NULL; 2899 if (cpu_buffer->commit_page == cpu_buffer->reader_page) 2900 goto out; 2901 2902 /* 2903 * Reset the reader page to size zero. 2904 */ 2905 local_set(&cpu_buffer->reader_page->write, 0); 2906 local_set(&cpu_buffer->reader_page->entries, 0); 2907 local_set(&cpu_buffer->reader_page->page->commit, 0); 2908 cpu_buffer->reader_page->real_end = 0; 2909 2910 spin: 2911 /* 2912 * Splice the empty reader page into the list around the head. 2913 */ 2914 reader = rb_set_head_page(cpu_buffer); 2915 cpu_buffer->reader_page->list.next = rb_list_head(reader->list.next); 2916 cpu_buffer->reader_page->list.prev = reader->list.prev; 2917 2918 /* 2919 * cpu_buffer->pages just needs to point to the buffer, it 2920 * has no specific buffer page to point to. Lets move it out 2921 * of our way so we don't accidently swap it. 2922 */ 2923 cpu_buffer->pages = reader->list.prev; 2924 2925 /* The reader page will be pointing to the new head */ 2926 rb_set_list_to_head(cpu_buffer, &cpu_buffer->reader_page->list); 2927 2928 /* 2929 * We want to make sure we read the overruns after we set up our 2930 * pointers to the next object. The writer side does a 2931 * cmpxchg to cross pages which acts as the mb on the writer 2932 * side. Note, the reader will constantly fail the swap 2933 * while the writer is updating the pointers, so this 2934 * guarantees that the overwrite recorded here is the one we 2935 * want to compare with the last_overrun. 2936 */ 2937 smp_mb(); 2938 overwrite = local_read(&(cpu_buffer->overrun)); 2939 2940 /* 2941 * Here's the tricky part. 2942 * 2943 * We need to move the pointer past the header page. 2944 * But we can only do that if a writer is not currently 2945 * moving it. The page before the header page has the 2946 * flag bit '1' set if it is pointing to the page we want. 2947 * but if the writer is in the process of moving it 2948 * than it will be '2' or already moved '0'. 2949 */ 2950 2951 ret = rb_head_page_replace(reader, cpu_buffer->reader_page); 2952 2953 /* 2954 * If we did not convert it, then we must try again. 2955 */ 2956 if (!ret) 2957 goto spin; 2958 2959 /* 2960 * Yeah! We succeeded in replacing the page. 2961 * 2962 * Now make the new head point back to the reader page. 2963 */ 2964 rb_list_head(reader->list.next)->prev = &cpu_buffer->reader_page->list; 2965 rb_inc_page(cpu_buffer, &cpu_buffer->head_page); 2966 2967 /* Finally update the reader page to the new head */ 2968 cpu_buffer->reader_page = reader; 2969 rb_reset_reader_page(cpu_buffer); 2970 2971 if (overwrite != cpu_buffer->last_overrun) { 2972 cpu_buffer->lost_events = overwrite - cpu_buffer->last_overrun; 2973 cpu_buffer->last_overrun = overwrite; 2974 } 2975 2976 goto again; 2977 2978 out: 2979 arch_spin_unlock(&cpu_buffer->lock); 2980 local_irq_restore(flags); 2981 2982 return reader; 2983 } 2984 2985 static void rb_advance_reader(struct ring_buffer_per_cpu *cpu_buffer) 2986 { 2987 struct ring_buffer_event *event; 2988 struct buffer_page *reader; 2989 unsigned length; 2990 2991 reader = rb_get_reader_page(cpu_buffer); 2992 2993 /* This function should not be called when buffer is empty */ 2994 if (RB_WARN_ON(cpu_buffer, !reader)) 2995 return; 2996 2997 event = rb_reader_event(cpu_buffer); 2998 2999 if (event->type_len <= RINGBUF_TYPE_DATA_TYPE_LEN_MAX) 3000 cpu_buffer->read++; 3001 3002 rb_update_read_stamp(cpu_buffer, event); 3003 3004 length = rb_event_length(event); 3005 cpu_buffer->reader_page->read += length; 3006 } 3007 3008 static void rb_advance_iter(struct ring_buffer_iter *iter) 3009 { 3010 struct ring_buffer *buffer; 3011 struct ring_buffer_per_cpu *cpu_buffer; 3012 struct ring_buffer_event *event; 3013 unsigned length; 3014 3015 cpu_buffer = iter->cpu_buffer; 3016 buffer = cpu_buffer->buffer; 3017 3018 /* 3019 * Check if we are at the end of the buffer. 3020 */ 3021 if (iter->head >= rb_page_size(iter->head_page)) { 3022 /* discarded commits can make the page empty */ 3023 if (iter->head_page == cpu_buffer->commit_page) 3024 return; 3025 rb_inc_iter(iter); 3026 return; 3027 } 3028 3029 event = rb_iter_head_event(iter); 3030 3031 length = rb_event_length(event); 3032 3033 /* 3034 * This should not be called to advance the header if we are 3035 * at the tail of the buffer. 3036 */ 3037 if (RB_WARN_ON(cpu_buffer, 3038 (iter->head_page == cpu_buffer->commit_page) && 3039 (iter->head + length > rb_commit_index(cpu_buffer)))) 3040 return; 3041 3042 rb_update_iter_read_stamp(iter, event); 3043 3044 iter->head += length; 3045 3046 /* check for end of page padding */ 3047 if ((iter->head >= rb_page_size(iter->head_page)) && 3048 (iter->head_page != cpu_buffer->commit_page)) 3049 rb_advance_iter(iter); 3050 } 3051 3052 static int rb_lost_events(struct ring_buffer_per_cpu *cpu_buffer) 3053 { 3054 return cpu_buffer->lost_events; 3055 } 3056 3057 static struct ring_buffer_event * 3058 rb_buffer_peek(struct ring_buffer_per_cpu *cpu_buffer, u64 *ts, 3059 unsigned long *lost_events) 3060 { 3061 struct ring_buffer_event *event; 3062 struct buffer_page *reader; 3063 int nr_loops = 0; 3064 3065 again: 3066 /* 3067 * We repeat when a timestamp is encountered. It is possible 3068 * to get multiple timestamps from an interrupt entering just 3069 * as one timestamp is about to be written, or from discarded 3070 * commits. The most that we can have is the number on a single page. 3071 */ 3072 if (RB_WARN_ON(cpu_buffer, ++nr_loops > RB_TIMESTAMPS_PER_PAGE)) 3073 return NULL; 3074 3075 reader = rb_get_reader_page(cpu_buffer); 3076 if (!reader) 3077 return NULL; 3078 3079 event = rb_reader_event(cpu_buffer); 3080 3081 switch (event->type_len) { 3082 case RINGBUF_TYPE_PADDING: 3083 if (rb_null_event(event)) 3084 RB_WARN_ON(cpu_buffer, 1); 3085 /* 3086 * Because the writer could be discarding every 3087 * event it creates (which would probably be bad) 3088 * if we were to go back to "again" then we may never 3089 * catch up, and will trigger the warn on, or lock 3090 * the box. Return the padding, and we will release 3091 * the current locks, and try again. 3092 */ 3093 return event; 3094 3095 case RINGBUF_TYPE_TIME_EXTEND: 3096 /* Internal data, OK to advance */ 3097 rb_advance_reader(cpu_buffer); 3098 goto again; 3099 3100 case RINGBUF_TYPE_TIME_STAMP: 3101 /* FIXME: not implemented */ 3102 rb_advance_reader(cpu_buffer); 3103 goto again; 3104 3105 case RINGBUF_TYPE_DATA: 3106 if (ts) { 3107 *ts = cpu_buffer->read_stamp + event->time_delta; 3108 ring_buffer_normalize_time_stamp(cpu_buffer->buffer, 3109 cpu_buffer->cpu, ts); 3110 } 3111 if (lost_events) 3112 *lost_events = rb_lost_events(cpu_buffer); 3113 return event; 3114 3115 default: 3116 BUG(); 3117 } 3118 3119 return NULL; 3120 } 3121 EXPORT_SYMBOL_GPL(ring_buffer_peek); 3122 3123 static struct ring_buffer_event * 3124 rb_iter_peek(struct ring_buffer_iter *iter, u64 *ts) 3125 { 3126 struct ring_buffer *buffer; 3127 struct ring_buffer_per_cpu *cpu_buffer; 3128 struct ring_buffer_event *event; 3129 int nr_loops = 0; 3130 3131 cpu_buffer = iter->cpu_buffer; 3132 buffer = cpu_buffer->buffer; 3133 3134 /* 3135 * Check if someone performed a consuming read to 3136 * the buffer. A consuming read invalidates the iterator 3137 * and we need to reset the iterator in this case. 3138 */ 3139 if (unlikely(iter->cache_read != cpu_buffer->read || 3140 iter->cache_reader_page != cpu_buffer->reader_page)) 3141 rb_iter_reset(iter); 3142 3143 again: 3144 if (ring_buffer_iter_empty(iter)) 3145 return NULL; 3146 3147 /* 3148 * We repeat when a timestamp is encountered. 3149 * We can get multiple timestamps by nested interrupts or also 3150 * if filtering is on (discarding commits). Since discarding 3151 * commits can be frequent we can get a lot of timestamps. 3152 * But we limit them by not adding timestamps if they begin 3153 * at the start of a page. 3154 */ 3155 if (RB_WARN_ON(cpu_buffer, ++nr_loops > RB_TIMESTAMPS_PER_PAGE)) 3156 return NULL; 3157 3158 if (rb_per_cpu_empty(cpu_buffer)) 3159 return NULL; 3160 3161 if (iter->head >= local_read(&iter->head_page->page->commit)) { 3162 rb_inc_iter(iter); 3163 goto again; 3164 } 3165 3166 event = rb_iter_head_event(iter); 3167 3168 switch (event->type_len) { 3169 case RINGBUF_TYPE_PADDING: 3170 if (rb_null_event(event)) { 3171 rb_inc_iter(iter); 3172 goto again; 3173 } 3174 rb_advance_iter(iter); 3175 return event; 3176 3177 case RINGBUF_TYPE_TIME_EXTEND: 3178 /* Internal data, OK to advance */ 3179 rb_advance_iter(iter); 3180 goto again; 3181 3182 case RINGBUF_TYPE_TIME_STAMP: 3183 /* FIXME: not implemented */ 3184 rb_advance_iter(iter); 3185 goto again; 3186 3187 case RINGBUF_TYPE_DATA: 3188 if (ts) { 3189 *ts = iter->read_stamp + event->time_delta; 3190 ring_buffer_normalize_time_stamp(buffer, 3191 cpu_buffer->cpu, ts); 3192 } 3193 return event; 3194 3195 default: 3196 BUG(); 3197 } 3198 3199 return NULL; 3200 } 3201 EXPORT_SYMBOL_GPL(ring_buffer_iter_peek); 3202 3203 static inline int rb_ok_to_lock(void) 3204 { 3205 /* 3206 * If an NMI die dumps out the content of the ring buffer 3207 * do not grab locks. We also permanently disable the ring 3208 * buffer too. A one time deal is all you get from reading 3209 * the ring buffer from an NMI. 3210 */ 3211 if (likely(!in_nmi())) 3212 return 1; 3213 3214 tracing_off_permanent(); 3215 return 0; 3216 } 3217 3218 /** 3219 * ring_buffer_peek - peek at the next event to be read 3220 * @buffer: The ring buffer to read 3221 * @cpu: The cpu to peak at 3222 * @ts: The timestamp counter of this event. 3223 * @lost_events: a variable to store if events were lost (may be NULL) 3224 * 3225 * This will return the event that will be read next, but does 3226 * not consume the data. 3227 */ 3228 struct ring_buffer_event * 3229 ring_buffer_peek(struct ring_buffer *buffer, int cpu, u64 *ts, 3230 unsigned long *lost_events) 3231 { 3232 struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu]; 3233 struct ring_buffer_event *event; 3234 unsigned long flags; 3235 int dolock; 3236 3237 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 3238 return NULL; 3239 3240 dolock = rb_ok_to_lock(); 3241 again: 3242 local_irq_save(flags); 3243 if (dolock) 3244 spin_lock(&cpu_buffer->reader_lock); 3245 event = rb_buffer_peek(cpu_buffer, ts, lost_events); 3246 if (event && event->type_len == RINGBUF_TYPE_PADDING) 3247 rb_advance_reader(cpu_buffer); 3248 if (dolock) 3249 spin_unlock(&cpu_buffer->reader_lock); 3250 local_irq_restore(flags); 3251 3252 if (event && event->type_len == RINGBUF_TYPE_PADDING) 3253 goto again; 3254 3255 return event; 3256 } 3257 3258 /** 3259 * ring_buffer_iter_peek - peek at the next event to be read 3260 * @iter: The ring buffer iterator 3261 * @ts: The timestamp counter of this event. 3262 * 3263 * This will return the event that will be read next, but does 3264 * not increment the iterator. 3265 */ 3266 struct ring_buffer_event * 3267 ring_buffer_iter_peek(struct ring_buffer_iter *iter, u64 *ts) 3268 { 3269 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer; 3270 struct ring_buffer_event *event; 3271 unsigned long flags; 3272 3273 again: 3274 spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 3275 event = rb_iter_peek(iter, ts); 3276 spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 3277 3278 if (event && event->type_len == RINGBUF_TYPE_PADDING) 3279 goto again; 3280 3281 return event; 3282 } 3283 3284 /** 3285 * ring_buffer_consume - return an event and consume it 3286 * @buffer: The ring buffer to get the next event from 3287 * @cpu: the cpu to read the buffer from 3288 * @ts: a variable to store the timestamp (may be NULL) 3289 * @lost_events: a variable to store if events were lost (may be NULL) 3290 * 3291 * Returns the next event in the ring buffer, and that event is consumed. 3292 * Meaning, that sequential reads will keep returning a different event, 3293 * and eventually empty the ring buffer if the producer is slower. 3294 */ 3295 struct ring_buffer_event * 3296 ring_buffer_consume(struct ring_buffer *buffer, int cpu, u64 *ts, 3297 unsigned long *lost_events) 3298 { 3299 struct ring_buffer_per_cpu *cpu_buffer; 3300 struct ring_buffer_event *event = NULL; 3301 unsigned long flags; 3302 int dolock; 3303 3304 dolock = rb_ok_to_lock(); 3305 3306 again: 3307 /* might be called in atomic */ 3308 preempt_disable(); 3309 3310 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 3311 goto out; 3312 3313 cpu_buffer = buffer->buffers[cpu]; 3314 local_irq_save(flags); 3315 if (dolock) 3316 spin_lock(&cpu_buffer->reader_lock); 3317 3318 event = rb_buffer_peek(cpu_buffer, ts, lost_events); 3319 if (event) { 3320 cpu_buffer->lost_events = 0; 3321 rb_advance_reader(cpu_buffer); 3322 } 3323 3324 if (dolock) 3325 spin_unlock(&cpu_buffer->reader_lock); 3326 local_irq_restore(flags); 3327 3328 out: 3329 preempt_enable(); 3330 3331 if (event && event->type_len == RINGBUF_TYPE_PADDING) 3332 goto again; 3333 3334 return event; 3335 } 3336 EXPORT_SYMBOL_GPL(ring_buffer_consume); 3337 3338 /** 3339 * ring_buffer_read_prepare - Prepare for a non consuming read of the buffer 3340 * @buffer: The ring buffer to read from 3341 * @cpu: The cpu buffer to iterate over 3342 * 3343 * This performs the initial preparations necessary to iterate 3344 * through the buffer. Memory is allocated, buffer recording 3345 * is disabled, and the iterator pointer is returned to the caller. 3346 * 3347 * Disabling buffer recordng prevents the reading from being 3348 * corrupted. This is not a consuming read, so a producer is not 3349 * expected. 3350 * 3351 * After a sequence of ring_buffer_read_prepare calls, the user is 3352 * expected to make at least one call to ring_buffer_prepare_sync. 3353 * Afterwards, ring_buffer_read_start is invoked to get things going 3354 * for real. 3355 * 3356 * This overall must be paired with ring_buffer_finish. 3357 */ 3358 struct ring_buffer_iter * 3359 ring_buffer_read_prepare(struct ring_buffer *buffer, int cpu) 3360 { 3361 struct ring_buffer_per_cpu *cpu_buffer; 3362 struct ring_buffer_iter *iter; 3363 3364 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 3365 return NULL; 3366 3367 iter = kmalloc(sizeof(*iter), GFP_KERNEL); 3368 if (!iter) 3369 return NULL; 3370 3371 cpu_buffer = buffer->buffers[cpu]; 3372 3373 iter->cpu_buffer = cpu_buffer; 3374 3375 atomic_inc(&cpu_buffer->record_disabled); 3376 3377 return iter; 3378 } 3379 EXPORT_SYMBOL_GPL(ring_buffer_read_prepare); 3380 3381 /** 3382 * ring_buffer_read_prepare_sync - Synchronize a set of prepare calls 3383 * 3384 * All previously invoked ring_buffer_read_prepare calls to prepare 3385 * iterators will be synchronized. Afterwards, read_buffer_read_start 3386 * calls on those iterators are allowed. 3387 */ 3388 void 3389 ring_buffer_read_prepare_sync(void) 3390 { 3391 synchronize_sched(); 3392 } 3393 EXPORT_SYMBOL_GPL(ring_buffer_read_prepare_sync); 3394 3395 /** 3396 * ring_buffer_read_start - start a non consuming read of the buffer 3397 * @iter: The iterator returned by ring_buffer_read_prepare 3398 * 3399 * This finalizes the startup of an iteration through the buffer. 3400 * The iterator comes from a call to ring_buffer_read_prepare and 3401 * an intervening ring_buffer_read_prepare_sync must have been 3402 * performed. 3403 * 3404 * Must be paired with ring_buffer_finish. 3405 */ 3406 void 3407 ring_buffer_read_start(struct ring_buffer_iter *iter) 3408 { 3409 struct ring_buffer_per_cpu *cpu_buffer; 3410 unsigned long flags; 3411 3412 if (!iter) 3413 return; 3414 3415 cpu_buffer = iter->cpu_buffer; 3416 3417 spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 3418 arch_spin_lock(&cpu_buffer->lock); 3419 rb_iter_reset(iter); 3420 arch_spin_unlock(&cpu_buffer->lock); 3421 spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 3422 } 3423 EXPORT_SYMBOL_GPL(ring_buffer_read_start); 3424 3425 /** 3426 * ring_buffer_finish - finish reading the iterator of the buffer 3427 * @iter: The iterator retrieved by ring_buffer_start 3428 * 3429 * This re-enables the recording to the buffer, and frees the 3430 * iterator. 3431 */ 3432 void 3433 ring_buffer_read_finish(struct ring_buffer_iter *iter) 3434 { 3435 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer; 3436 3437 atomic_dec(&cpu_buffer->record_disabled); 3438 kfree(iter); 3439 } 3440 EXPORT_SYMBOL_GPL(ring_buffer_read_finish); 3441 3442 /** 3443 * ring_buffer_read - read the next item in the ring buffer by the iterator 3444 * @iter: The ring buffer iterator 3445 * @ts: The time stamp of the event read. 3446 * 3447 * This reads the next event in the ring buffer and increments the iterator. 3448 */ 3449 struct ring_buffer_event * 3450 ring_buffer_read(struct ring_buffer_iter *iter, u64 *ts) 3451 { 3452 struct ring_buffer_event *event; 3453 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer; 3454 unsigned long flags; 3455 3456 spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 3457 again: 3458 event = rb_iter_peek(iter, ts); 3459 if (!event) 3460 goto out; 3461 3462 if (event->type_len == RINGBUF_TYPE_PADDING) 3463 goto again; 3464 3465 rb_advance_iter(iter); 3466 out: 3467 spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 3468 3469 return event; 3470 } 3471 EXPORT_SYMBOL_GPL(ring_buffer_read); 3472 3473 /** 3474 * ring_buffer_size - return the size of the ring buffer (in bytes) 3475 * @buffer: The ring buffer. 3476 */ 3477 unsigned long ring_buffer_size(struct ring_buffer *buffer) 3478 { 3479 return BUF_PAGE_SIZE * buffer->pages; 3480 } 3481 EXPORT_SYMBOL_GPL(ring_buffer_size); 3482 3483 static void 3484 rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer) 3485 { 3486 rb_head_page_deactivate(cpu_buffer); 3487 3488 cpu_buffer->head_page 3489 = list_entry(cpu_buffer->pages, struct buffer_page, list); 3490 local_set(&cpu_buffer->head_page->write, 0); 3491 local_set(&cpu_buffer->head_page->entries, 0); 3492 local_set(&cpu_buffer->head_page->page->commit, 0); 3493 3494 cpu_buffer->head_page->read = 0; 3495 3496 cpu_buffer->tail_page = cpu_buffer->head_page; 3497 cpu_buffer->commit_page = cpu_buffer->head_page; 3498 3499 INIT_LIST_HEAD(&cpu_buffer->reader_page->list); 3500 local_set(&cpu_buffer->reader_page->write, 0); 3501 local_set(&cpu_buffer->reader_page->entries, 0); 3502 local_set(&cpu_buffer->reader_page->page->commit, 0); 3503 cpu_buffer->reader_page->read = 0; 3504 3505 local_set(&cpu_buffer->commit_overrun, 0); 3506 local_set(&cpu_buffer->overrun, 0); 3507 local_set(&cpu_buffer->entries, 0); 3508 local_set(&cpu_buffer->committing, 0); 3509 local_set(&cpu_buffer->commits, 0); 3510 cpu_buffer->read = 0; 3511 3512 cpu_buffer->write_stamp = 0; 3513 cpu_buffer->read_stamp = 0; 3514 3515 cpu_buffer->lost_events = 0; 3516 cpu_buffer->last_overrun = 0; 3517 3518 rb_head_page_activate(cpu_buffer); 3519 } 3520 3521 /** 3522 * ring_buffer_reset_cpu - reset a ring buffer per CPU buffer 3523 * @buffer: The ring buffer to reset a per cpu buffer of 3524 * @cpu: The CPU buffer to be reset 3525 */ 3526 void ring_buffer_reset_cpu(struct ring_buffer *buffer, int cpu) 3527 { 3528 struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu]; 3529 unsigned long flags; 3530 3531 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 3532 return; 3533 3534 atomic_inc(&cpu_buffer->record_disabled); 3535 3536 spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 3537 3538 if (RB_WARN_ON(cpu_buffer, local_read(&cpu_buffer->committing))) 3539 goto out; 3540 3541 arch_spin_lock(&cpu_buffer->lock); 3542 3543 rb_reset_cpu(cpu_buffer); 3544 3545 arch_spin_unlock(&cpu_buffer->lock); 3546 3547 out: 3548 spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 3549 3550 atomic_dec(&cpu_buffer->record_disabled); 3551 } 3552 EXPORT_SYMBOL_GPL(ring_buffer_reset_cpu); 3553 3554 /** 3555 * ring_buffer_reset - reset a ring buffer 3556 * @buffer: The ring buffer to reset all cpu buffers 3557 */ 3558 void ring_buffer_reset(struct ring_buffer *buffer) 3559 { 3560 int cpu; 3561 3562 for_each_buffer_cpu(buffer, cpu) 3563 ring_buffer_reset_cpu(buffer, cpu); 3564 } 3565 EXPORT_SYMBOL_GPL(ring_buffer_reset); 3566 3567 /** 3568 * rind_buffer_empty - is the ring buffer empty? 3569 * @buffer: The ring buffer to test 3570 */ 3571 int ring_buffer_empty(struct ring_buffer *buffer) 3572 { 3573 struct ring_buffer_per_cpu *cpu_buffer; 3574 unsigned long flags; 3575 int dolock; 3576 int cpu; 3577 int ret; 3578 3579 dolock = rb_ok_to_lock(); 3580 3581 /* yes this is racy, but if you don't like the race, lock the buffer */ 3582 for_each_buffer_cpu(buffer, cpu) { 3583 cpu_buffer = buffer->buffers[cpu]; 3584 local_irq_save(flags); 3585 if (dolock) 3586 spin_lock(&cpu_buffer->reader_lock); 3587 ret = rb_per_cpu_empty(cpu_buffer); 3588 if (dolock) 3589 spin_unlock(&cpu_buffer->reader_lock); 3590 local_irq_restore(flags); 3591 3592 if (!ret) 3593 return 0; 3594 } 3595 3596 return 1; 3597 } 3598 EXPORT_SYMBOL_GPL(ring_buffer_empty); 3599 3600 /** 3601 * ring_buffer_empty_cpu - is a cpu buffer of a ring buffer empty? 3602 * @buffer: The ring buffer 3603 * @cpu: The CPU buffer to test 3604 */ 3605 int ring_buffer_empty_cpu(struct ring_buffer *buffer, int cpu) 3606 { 3607 struct ring_buffer_per_cpu *cpu_buffer; 3608 unsigned long flags; 3609 int dolock; 3610 int ret; 3611 3612 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 3613 return 1; 3614 3615 dolock = rb_ok_to_lock(); 3616 3617 cpu_buffer = buffer->buffers[cpu]; 3618 local_irq_save(flags); 3619 if (dolock) 3620 spin_lock(&cpu_buffer->reader_lock); 3621 ret = rb_per_cpu_empty(cpu_buffer); 3622 if (dolock) 3623 spin_unlock(&cpu_buffer->reader_lock); 3624 local_irq_restore(flags); 3625 3626 return ret; 3627 } 3628 EXPORT_SYMBOL_GPL(ring_buffer_empty_cpu); 3629 3630 #ifdef CONFIG_RING_BUFFER_ALLOW_SWAP 3631 /** 3632 * ring_buffer_swap_cpu - swap a CPU buffer between two ring buffers 3633 * @buffer_a: One buffer to swap with 3634 * @buffer_b: The other buffer to swap with 3635 * 3636 * This function is useful for tracers that want to take a "snapshot" 3637 * of a CPU buffer and has another back up buffer lying around. 3638 * it is expected that the tracer handles the cpu buffer not being 3639 * used at the moment. 3640 */ 3641 int ring_buffer_swap_cpu(struct ring_buffer *buffer_a, 3642 struct ring_buffer *buffer_b, int cpu) 3643 { 3644 struct ring_buffer_per_cpu *cpu_buffer_a; 3645 struct ring_buffer_per_cpu *cpu_buffer_b; 3646 int ret = -EINVAL; 3647 3648 if (!cpumask_test_cpu(cpu, buffer_a->cpumask) || 3649 !cpumask_test_cpu(cpu, buffer_b->cpumask)) 3650 goto out; 3651 3652 /* At least make sure the two buffers are somewhat the same */ 3653 if (buffer_a->pages != buffer_b->pages) 3654 goto out; 3655 3656 ret = -EAGAIN; 3657 3658 if (ring_buffer_flags != RB_BUFFERS_ON) 3659 goto out; 3660 3661 if (atomic_read(&buffer_a->record_disabled)) 3662 goto out; 3663 3664 if (atomic_read(&buffer_b->record_disabled)) 3665 goto out; 3666 3667 cpu_buffer_a = buffer_a->buffers[cpu]; 3668 cpu_buffer_b = buffer_b->buffers[cpu]; 3669 3670 if (atomic_read(&cpu_buffer_a->record_disabled)) 3671 goto out; 3672 3673 if (atomic_read(&cpu_buffer_b->record_disabled)) 3674 goto out; 3675 3676 /* 3677 * We can't do a synchronize_sched here because this 3678 * function can be called in atomic context. 3679 * Normally this will be called from the same CPU as cpu. 3680 * If not it's up to the caller to protect this. 3681 */ 3682 atomic_inc(&cpu_buffer_a->record_disabled); 3683 atomic_inc(&cpu_buffer_b->record_disabled); 3684 3685 ret = -EBUSY; 3686 if (local_read(&cpu_buffer_a->committing)) 3687 goto out_dec; 3688 if (local_read(&cpu_buffer_b->committing)) 3689 goto out_dec; 3690 3691 buffer_a->buffers[cpu] = cpu_buffer_b; 3692 buffer_b->buffers[cpu] = cpu_buffer_a; 3693 3694 cpu_buffer_b->buffer = buffer_a; 3695 cpu_buffer_a->buffer = buffer_b; 3696 3697 ret = 0; 3698 3699 out_dec: 3700 atomic_dec(&cpu_buffer_a->record_disabled); 3701 atomic_dec(&cpu_buffer_b->record_disabled); 3702 out: 3703 return ret; 3704 } 3705 EXPORT_SYMBOL_GPL(ring_buffer_swap_cpu); 3706 #endif /* CONFIG_RING_BUFFER_ALLOW_SWAP */ 3707 3708 /** 3709 * ring_buffer_alloc_read_page - allocate a page to read from buffer 3710 * @buffer: the buffer to allocate for. 3711 * 3712 * This function is used in conjunction with ring_buffer_read_page. 3713 * When reading a full page from the ring buffer, these functions 3714 * can be used to speed up the process. The calling function should 3715 * allocate a few pages first with this function. Then when it 3716 * needs to get pages from the ring buffer, it passes the result 3717 * of this function into ring_buffer_read_page, which will swap 3718 * the page that was allocated, with the read page of the buffer. 3719 * 3720 * Returns: 3721 * The page allocated, or NULL on error. 3722 */ 3723 void *ring_buffer_alloc_read_page(struct ring_buffer *buffer) 3724 { 3725 struct buffer_data_page *bpage; 3726 unsigned long addr; 3727 3728 addr = __get_free_page(GFP_KERNEL); 3729 if (!addr) 3730 return NULL; 3731 3732 bpage = (void *)addr; 3733 3734 rb_init_page(bpage); 3735 3736 return bpage; 3737 } 3738 EXPORT_SYMBOL_GPL(ring_buffer_alloc_read_page); 3739 3740 /** 3741 * ring_buffer_free_read_page - free an allocated read page 3742 * @buffer: the buffer the page was allocate for 3743 * @data: the page to free 3744 * 3745 * Free a page allocated from ring_buffer_alloc_read_page. 3746 */ 3747 void ring_buffer_free_read_page(struct ring_buffer *buffer, void *data) 3748 { 3749 free_page((unsigned long)data); 3750 } 3751 EXPORT_SYMBOL_GPL(ring_buffer_free_read_page); 3752 3753 /** 3754 * ring_buffer_read_page - extract a page from the ring buffer 3755 * @buffer: buffer to extract from 3756 * @data_page: the page to use allocated from ring_buffer_alloc_read_page 3757 * @len: amount to extract 3758 * @cpu: the cpu of the buffer to extract 3759 * @full: should the extraction only happen when the page is full. 3760 * 3761 * This function will pull out a page from the ring buffer and consume it. 3762 * @data_page must be the address of the variable that was returned 3763 * from ring_buffer_alloc_read_page. This is because the page might be used 3764 * to swap with a page in the ring buffer. 3765 * 3766 * for example: 3767 * rpage = ring_buffer_alloc_read_page(buffer); 3768 * if (!rpage) 3769 * return error; 3770 * ret = ring_buffer_read_page(buffer, &rpage, len, cpu, 0); 3771 * if (ret >= 0) 3772 * process_page(rpage, ret); 3773 * 3774 * When @full is set, the function will not return true unless 3775 * the writer is off the reader page. 3776 * 3777 * Note: it is up to the calling functions to handle sleeps and wakeups. 3778 * The ring buffer can be used anywhere in the kernel and can not 3779 * blindly call wake_up. The layer that uses the ring buffer must be 3780 * responsible for that. 3781 * 3782 * Returns: 3783 * >=0 if data has been transferred, returns the offset of consumed data. 3784 * <0 if no data has been transferred. 3785 */ 3786 int ring_buffer_read_page(struct ring_buffer *buffer, 3787 void **data_page, size_t len, int cpu, int full) 3788 { 3789 struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu]; 3790 struct ring_buffer_event *event; 3791 struct buffer_data_page *bpage; 3792 struct buffer_page *reader; 3793 unsigned long missed_events; 3794 unsigned long flags; 3795 unsigned int commit; 3796 unsigned int read; 3797 u64 save_timestamp; 3798 int ret = -1; 3799 3800 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 3801 goto out; 3802 3803 /* 3804 * If len is not big enough to hold the page header, then 3805 * we can not copy anything. 3806 */ 3807 if (len <= BUF_PAGE_HDR_SIZE) 3808 goto out; 3809 3810 len -= BUF_PAGE_HDR_SIZE; 3811 3812 if (!data_page) 3813 goto out; 3814 3815 bpage = *data_page; 3816 if (!bpage) 3817 goto out; 3818 3819 spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 3820 3821 reader = rb_get_reader_page(cpu_buffer); 3822 if (!reader) 3823 goto out_unlock; 3824 3825 event = rb_reader_event(cpu_buffer); 3826 3827 read = reader->read; 3828 commit = rb_page_commit(reader); 3829 3830 /* Check if any events were dropped */ 3831 missed_events = cpu_buffer->lost_events; 3832 3833 /* 3834 * If this page has been partially read or 3835 * if len is not big enough to read the rest of the page or 3836 * a writer is still on the page, then 3837 * we must copy the data from the page to the buffer. 3838 * Otherwise, we can simply swap the page with the one passed in. 3839 */ 3840 if (read || (len < (commit - read)) || 3841 cpu_buffer->reader_page == cpu_buffer->commit_page) { 3842 struct buffer_data_page *rpage = cpu_buffer->reader_page->page; 3843 unsigned int rpos = read; 3844 unsigned int pos = 0; 3845 unsigned int size; 3846 3847 if (full) 3848 goto out_unlock; 3849 3850 if (len > (commit - read)) 3851 len = (commit - read); 3852 3853 size = rb_event_length(event); 3854 3855 if (len < size) 3856 goto out_unlock; 3857 3858 /* save the current timestamp, since the user will need it */ 3859 save_timestamp = cpu_buffer->read_stamp; 3860 3861 /* Need to copy one event at a time */ 3862 do { 3863 memcpy(bpage->data + pos, rpage->data + rpos, size); 3864 3865 len -= size; 3866 3867 rb_advance_reader(cpu_buffer); 3868 rpos = reader->read; 3869 pos += size; 3870 3871 event = rb_reader_event(cpu_buffer); 3872 size = rb_event_length(event); 3873 } while (len > size); 3874 3875 /* update bpage */ 3876 local_set(&bpage->commit, pos); 3877 bpage->time_stamp = save_timestamp; 3878 3879 /* we copied everything to the beginning */ 3880 read = 0; 3881 } else { 3882 /* update the entry counter */ 3883 cpu_buffer->read += rb_page_entries(reader); 3884 3885 /* swap the pages */ 3886 rb_init_page(bpage); 3887 bpage = reader->page; 3888 reader->page = *data_page; 3889 local_set(&reader->write, 0); 3890 local_set(&reader->entries, 0); 3891 reader->read = 0; 3892 *data_page = bpage; 3893 3894 /* 3895 * Use the real_end for the data size, 3896 * This gives us a chance to store the lost events 3897 * on the page. 3898 */ 3899 if (reader->real_end) 3900 local_set(&bpage->commit, reader->real_end); 3901 } 3902 ret = read; 3903 3904 cpu_buffer->lost_events = 0; 3905 3906 commit = local_read(&bpage->commit); 3907 /* 3908 * Set a flag in the commit field if we lost events 3909 */ 3910 if (missed_events) { 3911 /* If there is room at the end of the page to save the 3912 * missed events, then record it there. 3913 */ 3914 if (BUF_PAGE_SIZE - commit >= sizeof(missed_events)) { 3915 memcpy(&bpage->data[commit], &missed_events, 3916 sizeof(missed_events)); 3917 local_add(RB_MISSED_STORED, &bpage->commit); 3918 commit += sizeof(missed_events); 3919 } 3920 local_add(RB_MISSED_EVENTS, &bpage->commit); 3921 } 3922 3923 /* 3924 * This page may be off to user land. Zero it out here. 3925 */ 3926 if (commit < BUF_PAGE_SIZE) 3927 memset(&bpage->data[commit], 0, BUF_PAGE_SIZE - commit); 3928 3929 out_unlock: 3930 spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 3931 3932 out: 3933 return ret; 3934 } 3935 EXPORT_SYMBOL_GPL(ring_buffer_read_page); 3936 3937 #ifdef CONFIG_TRACING 3938 static ssize_t 3939 rb_simple_read(struct file *filp, char __user *ubuf, 3940 size_t cnt, loff_t *ppos) 3941 { 3942 unsigned long *p = filp->private_data; 3943 char buf[64]; 3944 int r; 3945 3946 if (test_bit(RB_BUFFERS_DISABLED_BIT, p)) 3947 r = sprintf(buf, "permanently disabled\n"); 3948 else 3949 r = sprintf(buf, "%d\n", test_bit(RB_BUFFERS_ON_BIT, p)); 3950 3951 return simple_read_from_buffer(ubuf, cnt, ppos, buf, r); 3952 } 3953 3954 static ssize_t 3955 rb_simple_write(struct file *filp, const char __user *ubuf, 3956 size_t cnt, loff_t *ppos) 3957 { 3958 unsigned long *p = filp->private_data; 3959 char buf[64]; 3960 unsigned long val; 3961 int ret; 3962 3963 if (cnt >= sizeof(buf)) 3964 return -EINVAL; 3965 3966 if (copy_from_user(&buf, ubuf, cnt)) 3967 return -EFAULT; 3968 3969 buf[cnt] = 0; 3970 3971 ret = strict_strtoul(buf, 10, &val); 3972 if (ret < 0) 3973 return ret; 3974 3975 if (val) 3976 set_bit(RB_BUFFERS_ON_BIT, p); 3977 else 3978 clear_bit(RB_BUFFERS_ON_BIT, p); 3979 3980 (*ppos)++; 3981 3982 return cnt; 3983 } 3984 3985 static const struct file_operations rb_simple_fops = { 3986 .open = tracing_open_generic, 3987 .read = rb_simple_read, 3988 .write = rb_simple_write, 3989 }; 3990 3991 3992 static __init int rb_init_debugfs(void) 3993 { 3994 struct dentry *d_tracer; 3995 3996 d_tracer = tracing_init_dentry(); 3997 3998 trace_create_file("tracing_on", 0644, d_tracer, 3999 &ring_buffer_flags, &rb_simple_fops); 4000 4001 return 0; 4002 } 4003 4004 fs_initcall(rb_init_debugfs); 4005 #endif 4006 4007 #ifdef CONFIG_HOTPLUG_CPU 4008 static int rb_cpu_notify(struct notifier_block *self, 4009 unsigned long action, void *hcpu) 4010 { 4011 struct ring_buffer *buffer = 4012 container_of(self, struct ring_buffer, cpu_notify); 4013 long cpu = (long)hcpu; 4014 4015 switch (action) { 4016 case CPU_UP_PREPARE: 4017 case CPU_UP_PREPARE_FROZEN: 4018 if (cpumask_test_cpu(cpu, buffer->cpumask)) 4019 return NOTIFY_OK; 4020 4021 buffer->buffers[cpu] = 4022 rb_allocate_cpu_buffer(buffer, cpu); 4023 if (!buffer->buffers[cpu]) { 4024 WARN(1, "failed to allocate ring buffer on CPU %ld\n", 4025 cpu); 4026 return NOTIFY_OK; 4027 } 4028 smp_wmb(); 4029 cpumask_set_cpu(cpu, buffer->cpumask); 4030 break; 4031 case CPU_DOWN_PREPARE: 4032 case CPU_DOWN_PREPARE_FROZEN: 4033 /* 4034 * Do nothing. 4035 * If we were to free the buffer, then the user would 4036 * lose any trace that was in the buffer. 4037 */ 4038 break; 4039 default: 4040 break; 4041 } 4042 return NOTIFY_OK; 4043 } 4044 #endif 4045