1 /* 2 * Generic ring buffer 3 * 4 * Copyright (C) 2008 Steven Rostedt <srostedt@redhat.com> 5 */ 6 #include <linux/ftrace_event.h> 7 #include <linux/ring_buffer.h> 8 #include <linux/trace_clock.h> 9 #include <linux/trace_seq.h> 10 #include <linux/spinlock.h> 11 #include <linux/irq_work.h> 12 #include <linux/debugfs.h> 13 #include <linux/uaccess.h> 14 #include <linux/hardirq.h> 15 #include <linux/kthread.h> /* for self test */ 16 #include <linux/kmemcheck.h> 17 #include <linux/module.h> 18 #include <linux/percpu.h> 19 #include <linux/mutex.h> 20 #include <linux/delay.h> 21 #include <linux/slab.h> 22 #include <linux/init.h> 23 #include <linux/hash.h> 24 #include <linux/list.h> 25 #include <linux/cpu.h> 26 #include <linux/fs.h> 27 28 #include <asm/local.h> 29 30 static void update_pages_handler(struct work_struct *work); 31 32 /* 33 * The ring buffer header is special. We must manually up keep it. 34 */ 35 int ring_buffer_print_entry_header(struct trace_seq *s) 36 { 37 trace_seq_puts(s, "# compressed entry header\n"); 38 trace_seq_puts(s, "\ttype_len : 5 bits\n"); 39 trace_seq_puts(s, "\ttime_delta : 27 bits\n"); 40 trace_seq_puts(s, "\tarray : 32 bits\n"); 41 trace_seq_putc(s, '\n'); 42 trace_seq_printf(s, "\tpadding : type == %d\n", 43 RINGBUF_TYPE_PADDING); 44 trace_seq_printf(s, "\ttime_extend : type == %d\n", 45 RINGBUF_TYPE_TIME_EXTEND); 46 trace_seq_printf(s, "\tdata max type_len == %d\n", 47 RINGBUF_TYPE_DATA_TYPE_LEN_MAX); 48 49 return !trace_seq_has_overflowed(s); 50 } 51 52 /* 53 * The ring buffer is made up of a list of pages. A separate list of pages is 54 * allocated for each CPU. A writer may only write to a buffer that is 55 * associated with the CPU it is currently executing on. A reader may read 56 * from any per cpu buffer. 57 * 58 * The reader is special. For each per cpu buffer, the reader has its own 59 * reader page. When a reader has read the entire reader page, this reader 60 * page is swapped with another page in the ring buffer. 61 * 62 * Now, as long as the writer is off the reader page, the reader can do what 63 * ever it wants with that page. The writer will never write to that page 64 * again (as long as it is out of the ring buffer). 65 * 66 * Here's some silly ASCII art. 67 * 68 * +------+ 69 * |reader| RING BUFFER 70 * |page | 71 * +------+ +---+ +---+ +---+ 72 * | |-->| |-->| | 73 * +---+ +---+ +---+ 74 * ^ | 75 * | | 76 * +---------------+ 77 * 78 * 79 * +------+ 80 * |reader| RING BUFFER 81 * |page |------------------v 82 * +------+ +---+ +---+ +---+ 83 * | |-->| |-->| | 84 * +---+ +---+ +---+ 85 * ^ | 86 * | | 87 * +---------------+ 88 * 89 * 90 * +------+ 91 * |reader| RING BUFFER 92 * |page |------------------v 93 * +------+ +---+ +---+ +---+ 94 * ^ | |-->| |-->| | 95 * | +---+ +---+ +---+ 96 * | | 97 * | | 98 * +------------------------------+ 99 * 100 * 101 * +------+ 102 * |buffer| RING BUFFER 103 * |page |------------------v 104 * +------+ +---+ +---+ +---+ 105 * ^ | | | |-->| | 106 * | New +---+ +---+ +---+ 107 * | Reader------^ | 108 * | page | 109 * +------------------------------+ 110 * 111 * 112 * After we make this swap, the reader can hand this page off to the splice 113 * code and be done with it. It can even allocate a new page if it needs to 114 * and swap that into the ring buffer. 115 * 116 * We will be using cmpxchg soon to make all this lockless. 117 * 118 */ 119 120 /* 121 * A fast way to enable or disable all ring buffers is to 122 * call tracing_on or tracing_off. Turning off the ring buffers 123 * prevents all ring buffers from being recorded to. 124 * Turning this switch on, makes it OK to write to the 125 * ring buffer, if the ring buffer is enabled itself. 126 * 127 * There's three layers that must be on in order to write 128 * to the ring buffer. 129 * 130 * 1) This global flag must be set. 131 * 2) The ring buffer must be enabled for recording. 132 * 3) The per cpu buffer must be enabled for recording. 133 * 134 * In case of an anomaly, this global flag has a bit set that 135 * will permantly disable all ring buffers. 136 */ 137 138 /* 139 * Global flag to disable all recording to ring buffers 140 * This has two bits: ON, DISABLED 141 * 142 * ON DISABLED 143 * ---- ---------- 144 * 0 0 : ring buffers are off 145 * 1 0 : ring buffers are on 146 * X 1 : ring buffers are permanently disabled 147 */ 148 149 enum { 150 RB_BUFFERS_ON_BIT = 0, 151 RB_BUFFERS_DISABLED_BIT = 1, 152 }; 153 154 enum { 155 RB_BUFFERS_ON = 1 << RB_BUFFERS_ON_BIT, 156 RB_BUFFERS_DISABLED = 1 << RB_BUFFERS_DISABLED_BIT, 157 }; 158 159 static unsigned long ring_buffer_flags __read_mostly = RB_BUFFERS_ON; 160 161 /* Used for individual buffers (after the counter) */ 162 #define RB_BUFFER_OFF (1 << 20) 163 164 #define BUF_PAGE_HDR_SIZE offsetof(struct buffer_data_page, data) 165 166 /** 167 * tracing_off_permanent - permanently disable ring buffers 168 * 169 * This function, once called, will disable all ring buffers 170 * permanently. 171 */ 172 void tracing_off_permanent(void) 173 { 174 set_bit(RB_BUFFERS_DISABLED_BIT, &ring_buffer_flags); 175 } 176 177 #define RB_EVNT_HDR_SIZE (offsetof(struct ring_buffer_event, array)) 178 #define RB_ALIGNMENT 4U 179 #define RB_MAX_SMALL_DATA (RB_ALIGNMENT * RINGBUF_TYPE_DATA_TYPE_LEN_MAX) 180 #define RB_EVNT_MIN_SIZE 8U /* two 32bit words */ 181 182 #ifndef CONFIG_HAVE_64BIT_ALIGNED_ACCESS 183 # define RB_FORCE_8BYTE_ALIGNMENT 0 184 # define RB_ARCH_ALIGNMENT RB_ALIGNMENT 185 #else 186 # define RB_FORCE_8BYTE_ALIGNMENT 1 187 # define RB_ARCH_ALIGNMENT 8U 188 #endif 189 190 #define RB_ALIGN_DATA __aligned(RB_ARCH_ALIGNMENT) 191 192 /* define RINGBUF_TYPE_DATA for 'case RINGBUF_TYPE_DATA:' */ 193 #define RINGBUF_TYPE_DATA 0 ... RINGBUF_TYPE_DATA_TYPE_LEN_MAX 194 195 enum { 196 RB_LEN_TIME_EXTEND = 8, 197 RB_LEN_TIME_STAMP = 16, 198 }; 199 200 #define skip_time_extend(event) \ 201 ((struct ring_buffer_event *)((char *)event + RB_LEN_TIME_EXTEND)) 202 203 static inline int rb_null_event(struct ring_buffer_event *event) 204 { 205 return event->type_len == RINGBUF_TYPE_PADDING && !event->time_delta; 206 } 207 208 static void rb_event_set_padding(struct ring_buffer_event *event) 209 { 210 /* padding has a NULL time_delta */ 211 event->type_len = RINGBUF_TYPE_PADDING; 212 event->time_delta = 0; 213 } 214 215 static unsigned 216 rb_event_data_length(struct ring_buffer_event *event) 217 { 218 unsigned length; 219 220 if (event->type_len) 221 length = event->type_len * RB_ALIGNMENT; 222 else 223 length = event->array[0]; 224 return length + RB_EVNT_HDR_SIZE; 225 } 226 227 /* 228 * Return the length of the given event. Will return 229 * the length of the time extend if the event is a 230 * time extend. 231 */ 232 static inline unsigned 233 rb_event_length(struct ring_buffer_event *event) 234 { 235 switch (event->type_len) { 236 case RINGBUF_TYPE_PADDING: 237 if (rb_null_event(event)) 238 /* undefined */ 239 return -1; 240 return event->array[0] + RB_EVNT_HDR_SIZE; 241 242 case RINGBUF_TYPE_TIME_EXTEND: 243 return RB_LEN_TIME_EXTEND; 244 245 case RINGBUF_TYPE_TIME_STAMP: 246 return RB_LEN_TIME_STAMP; 247 248 case RINGBUF_TYPE_DATA: 249 return rb_event_data_length(event); 250 default: 251 BUG(); 252 } 253 /* not hit */ 254 return 0; 255 } 256 257 /* 258 * Return total length of time extend and data, 259 * or just the event length for all other events. 260 */ 261 static inline unsigned 262 rb_event_ts_length(struct ring_buffer_event *event) 263 { 264 unsigned len = 0; 265 266 if (event->type_len == RINGBUF_TYPE_TIME_EXTEND) { 267 /* time extends include the data event after it */ 268 len = RB_LEN_TIME_EXTEND; 269 event = skip_time_extend(event); 270 } 271 return len + rb_event_length(event); 272 } 273 274 /** 275 * ring_buffer_event_length - return the length of the event 276 * @event: the event to get the length of 277 * 278 * Returns the size of the data load of a data event. 279 * If the event is something other than a data event, it 280 * returns the size of the event itself. With the exception 281 * of a TIME EXTEND, where it still returns the size of the 282 * data load of the data event after it. 283 */ 284 unsigned ring_buffer_event_length(struct ring_buffer_event *event) 285 { 286 unsigned length; 287 288 if (event->type_len == RINGBUF_TYPE_TIME_EXTEND) 289 event = skip_time_extend(event); 290 291 length = rb_event_length(event); 292 if (event->type_len > RINGBUF_TYPE_DATA_TYPE_LEN_MAX) 293 return length; 294 length -= RB_EVNT_HDR_SIZE; 295 if (length > RB_MAX_SMALL_DATA + sizeof(event->array[0])) 296 length -= sizeof(event->array[0]); 297 return length; 298 } 299 EXPORT_SYMBOL_GPL(ring_buffer_event_length); 300 301 /* inline for ring buffer fast paths */ 302 static void * 303 rb_event_data(struct ring_buffer_event *event) 304 { 305 if (event->type_len == RINGBUF_TYPE_TIME_EXTEND) 306 event = skip_time_extend(event); 307 BUG_ON(event->type_len > RINGBUF_TYPE_DATA_TYPE_LEN_MAX); 308 /* If length is in len field, then array[0] has the data */ 309 if (event->type_len) 310 return (void *)&event->array[0]; 311 /* Otherwise length is in array[0] and array[1] has the data */ 312 return (void *)&event->array[1]; 313 } 314 315 /** 316 * ring_buffer_event_data - return the data of the event 317 * @event: the event to get the data from 318 */ 319 void *ring_buffer_event_data(struct ring_buffer_event *event) 320 { 321 return rb_event_data(event); 322 } 323 EXPORT_SYMBOL_GPL(ring_buffer_event_data); 324 325 #define for_each_buffer_cpu(buffer, cpu) \ 326 for_each_cpu(cpu, buffer->cpumask) 327 328 #define TS_SHIFT 27 329 #define TS_MASK ((1ULL << TS_SHIFT) - 1) 330 #define TS_DELTA_TEST (~TS_MASK) 331 332 /* Flag when events were overwritten */ 333 #define RB_MISSED_EVENTS (1 << 31) 334 /* Missed count stored at end */ 335 #define RB_MISSED_STORED (1 << 30) 336 337 struct buffer_data_page { 338 u64 time_stamp; /* page time stamp */ 339 local_t commit; /* write committed index */ 340 unsigned char data[] RB_ALIGN_DATA; /* data of buffer page */ 341 }; 342 343 /* 344 * Note, the buffer_page list must be first. The buffer pages 345 * are allocated in cache lines, which means that each buffer 346 * page will be at the beginning of a cache line, and thus 347 * the least significant bits will be zero. We use this to 348 * add flags in the list struct pointers, to make the ring buffer 349 * lockless. 350 */ 351 struct buffer_page { 352 struct list_head list; /* list of buffer pages */ 353 local_t write; /* index for next write */ 354 unsigned read; /* index for next read */ 355 local_t entries; /* entries on this page */ 356 unsigned long real_end; /* real end of data */ 357 struct buffer_data_page *page; /* Actual data page */ 358 }; 359 360 /* 361 * The buffer page counters, write and entries, must be reset 362 * atomically when crossing page boundaries. To synchronize this 363 * update, two counters are inserted into the number. One is 364 * the actual counter for the write position or count on the page. 365 * 366 * The other is a counter of updaters. Before an update happens 367 * the update partition of the counter is incremented. This will 368 * allow the updater to update the counter atomically. 369 * 370 * The counter is 20 bits, and the state data is 12. 371 */ 372 #define RB_WRITE_MASK 0xfffff 373 #define RB_WRITE_INTCNT (1 << 20) 374 375 static void rb_init_page(struct buffer_data_page *bpage) 376 { 377 local_set(&bpage->commit, 0); 378 } 379 380 /** 381 * ring_buffer_page_len - the size of data on the page. 382 * @page: The page to read 383 * 384 * Returns the amount of data on the page, including buffer page header. 385 */ 386 size_t ring_buffer_page_len(void *page) 387 { 388 return local_read(&((struct buffer_data_page *)page)->commit) 389 + BUF_PAGE_HDR_SIZE; 390 } 391 392 /* 393 * Also stolen from mm/slob.c. Thanks to Mathieu Desnoyers for pointing 394 * this issue out. 395 */ 396 static void free_buffer_page(struct buffer_page *bpage) 397 { 398 free_page((unsigned long)bpage->page); 399 kfree(bpage); 400 } 401 402 /* 403 * We need to fit the time_stamp delta into 27 bits. 404 */ 405 static inline int test_time_stamp(u64 delta) 406 { 407 if (delta & TS_DELTA_TEST) 408 return 1; 409 return 0; 410 } 411 412 #define BUF_PAGE_SIZE (PAGE_SIZE - BUF_PAGE_HDR_SIZE) 413 414 /* Max payload is BUF_PAGE_SIZE - header (8bytes) */ 415 #define BUF_MAX_DATA_SIZE (BUF_PAGE_SIZE - (sizeof(u32) * 2)) 416 417 int ring_buffer_print_page_header(struct trace_seq *s) 418 { 419 struct buffer_data_page field; 420 421 trace_seq_printf(s, "\tfield: u64 timestamp;\t" 422 "offset:0;\tsize:%u;\tsigned:%u;\n", 423 (unsigned int)sizeof(field.time_stamp), 424 (unsigned int)is_signed_type(u64)); 425 426 trace_seq_printf(s, "\tfield: local_t commit;\t" 427 "offset:%u;\tsize:%u;\tsigned:%u;\n", 428 (unsigned int)offsetof(typeof(field), commit), 429 (unsigned int)sizeof(field.commit), 430 (unsigned int)is_signed_type(long)); 431 432 trace_seq_printf(s, "\tfield: int overwrite;\t" 433 "offset:%u;\tsize:%u;\tsigned:%u;\n", 434 (unsigned int)offsetof(typeof(field), commit), 435 1, 436 (unsigned int)is_signed_type(long)); 437 438 trace_seq_printf(s, "\tfield: char data;\t" 439 "offset:%u;\tsize:%u;\tsigned:%u;\n", 440 (unsigned int)offsetof(typeof(field), data), 441 (unsigned int)BUF_PAGE_SIZE, 442 (unsigned int)is_signed_type(char)); 443 444 return !trace_seq_has_overflowed(s); 445 } 446 447 struct rb_irq_work { 448 struct irq_work work; 449 wait_queue_head_t waiters; 450 bool waiters_pending; 451 }; 452 453 /* 454 * head_page == tail_page && head == tail then buffer is empty. 455 */ 456 struct ring_buffer_per_cpu { 457 int cpu; 458 atomic_t record_disabled; 459 struct ring_buffer *buffer; 460 raw_spinlock_t reader_lock; /* serialize readers */ 461 arch_spinlock_t lock; 462 struct lock_class_key lock_key; 463 unsigned int nr_pages; 464 struct list_head *pages; 465 struct buffer_page *head_page; /* read from head */ 466 struct buffer_page *tail_page; /* write to tail */ 467 struct buffer_page *commit_page; /* committed pages */ 468 struct buffer_page *reader_page; 469 unsigned long lost_events; 470 unsigned long last_overrun; 471 local_t entries_bytes; 472 local_t entries; 473 local_t overrun; 474 local_t commit_overrun; 475 local_t dropped_events; 476 local_t committing; 477 local_t commits; 478 unsigned long read; 479 unsigned long read_bytes; 480 u64 write_stamp; 481 u64 read_stamp; 482 /* ring buffer pages to update, > 0 to add, < 0 to remove */ 483 int nr_pages_to_update; 484 struct list_head new_pages; /* new pages to add */ 485 struct work_struct update_pages_work; 486 struct completion update_done; 487 488 struct rb_irq_work irq_work; 489 }; 490 491 struct ring_buffer { 492 unsigned flags; 493 int cpus; 494 atomic_t record_disabled; 495 atomic_t resize_disabled; 496 cpumask_var_t cpumask; 497 498 struct lock_class_key *reader_lock_key; 499 500 struct mutex mutex; 501 502 struct ring_buffer_per_cpu **buffers; 503 504 #ifdef CONFIG_HOTPLUG_CPU 505 struct notifier_block cpu_notify; 506 #endif 507 u64 (*clock)(void); 508 509 struct rb_irq_work irq_work; 510 }; 511 512 struct ring_buffer_iter { 513 struct ring_buffer_per_cpu *cpu_buffer; 514 unsigned long head; 515 struct buffer_page *head_page; 516 struct buffer_page *cache_reader_page; 517 unsigned long cache_read; 518 u64 read_stamp; 519 }; 520 521 /* 522 * rb_wake_up_waiters - wake up tasks waiting for ring buffer input 523 * 524 * Schedules a delayed work to wake up any task that is blocked on the 525 * ring buffer waiters queue. 526 */ 527 static void rb_wake_up_waiters(struct irq_work *work) 528 { 529 struct rb_irq_work *rbwork = container_of(work, struct rb_irq_work, work); 530 531 wake_up_all(&rbwork->waiters); 532 } 533 534 /** 535 * ring_buffer_wait - wait for input to the ring buffer 536 * @buffer: buffer to wait on 537 * @cpu: the cpu buffer to wait on 538 * @full: wait until a full page is available, if @cpu != RING_BUFFER_ALL_CPUS 539 * 540 * If @cpu == RING_BUFFER_ALL_CPUS then the task will wake up as soon 541 * as data is added to any of the @buffer's cpu buffers. Otherwise 542 * it will wait for data to be added to a specific cpu buffer. 543 */ 544 int ring_buffer_wait(struct ring_buffer *buffer, int cpu, bool full) 545 { 546 struct ring_buffer_per_cpu *uninitialized_var(cpu_buffer); 547 DEFINE_WAIT(wait); 548 struct rb_irq_work *work; 549 int ret = 0; 550 551 /* 552 * Depending on what the caller is waiting for, either any 553 * data in any cpu buffer, or a specific buffer, put the 554 * caller on the appropriate wait queue. 555 */ 556 if (cpu == RING_BUFFER_ALL_CPUS) 557 work = &buffer->irq_work; 558 else { 559 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 560 return -ENODEV; 561 cpu_buffer = buffer->buffers[cpu]; 562 work = &cpu_buffer->irq_work; 563 } 564 565 566 while (true) { 567 prepare_to_wait(&work->waiters, &wait, TASK_INTERRUPTIBLE); 568 569 /* 570 * The events can happen in critical sections where 571 * checking a work queue can cause deadlocks. 572 * After adding a task to the queue, this flag is set 573 * only to notify events to try to wake up the queue 574 * using irq_work. 575 * 576 * We don't clear it even if the buffer is no longer 577 * empty. The flag only causes the next event to run 578 * irq_work to do the work queue wake up. The worse 579 * that can happen if we race with !trace_empty() is that 580 * an event will cause an irq_work to try to wake up 581 * an empty queue. 582 * 583 * There's no reason to protect this flag either, as 584 * the work queue and irq_work logic will do the necessary 585 * synchronization for the wake ups. The only thing 586 * that is necessary is that the wake up happens after 587 * a task has been queued. It's OK for spurious wake ups. 588 */ 589 work->waiters_pending = true; 590 591 if (signal_pending(current)) { 592 ret = -EINTR; 593 break; 594 } 595 596 if (cpu == RING_BUFFER_ALL_CPUS && !ring_buffer_empty(buffer)) 597 break; 598 599 if (cpu != RING_BUFFER_ALL_CPUS && 600 !ring_buffer_empty_cpu(buffer, cpu)) { 601 unsigned long flags; 602 bool pagebusy; 603 604 if (!full) 605 break; 606 607 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 608 pagebusy = cpu_buffer->reader_page == cpu_buffer->commit_page; 609 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 610 611 if (!pagebusy) 612 break; 613 } 614 615 schedule(); 616 } 617 618 finish_wait(&work->waiters, &wait); 619 620 return ret; 621 } 622 623 /** 624 * ring_buffer_poll_wait - poll on buffer input 625 * @buffer: buffer to wait on 626 * @cpu: the cpu buffer to wait on 627 * @filp: the file descriptor 628 * @poll_table: The poll descriptor 629 * 630 * If @cpu == RING_BUFFER_ALL_CPUS then the task will wake up as soon 631 * as data is added to any of the @buffer's cpu buffers. Otherwise 632 * it will wait for data to be added to a specific cpu buffer. 633 * 634 * Returns POLLIN | POLLRDNORM if data exists in the buffers, 635 * zero otherwise. 636 */ 637 int ring_buffer_poll_wait(struct ring_buffer *buffer, int cpu, 638 struct file *filp, poll_table *poll_table) 639 { 640 struct ring_buffer_per_cpu *cpu_buffer; 641 struct rb_irq_work *work; 642 643 if (cpu == RING_BUFFER_ALL_CPUS) 644 work = &buffer->irq_work; 645 else { 646 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 647 return -EINVAL; 648 649 cpu_buffer = buffer->buffers[cpu]; 650 work = &cpu_buffer->irq_work; 651 } 652 653 poll_wait(filp, &work->waiters, poll_table); 654 work->waiters_pending = true; 655 /* 656 * There's a tight race between setting the waiters_pending and 657 * checking if the ring buffer is empty. Once the waiters_pending bit 658 * is set, the next event will wake the task up, but we can get stuck 659 * if there's only a single event in. 660 * 661 * FIXME: Ideally, we need a memory barrier on the writer side as well, 662 * but adding a memory barrier to all events will cause too much of a 663 * performance hit in the fast path. We only need a memory barrier when 664 * the buffer goes from empty to having content. But as this race is 665 * extremely small, and it's not a problem if another event comes in, we 666 * will fix it later. 667 */ 668 smp_mb(); 669 670 if ((cpu == RING_BUFFER_ALL_CPUS && !ring_buffer_empty(buffer)) || 671 (cpu != RING_BUFFER_ALL_CPUS && !ring_buffer_empty_cpu(buffer, cpu))) 672 return POLLIN | POLLRDNORM; 673 return 0; 674 } 675 676 /* buffer may be either ring_buffer or ring_buffer_per_cpu */ 677 #define RB_WARN_ON(b, cond) \ 678 ({ \ 679 int _____ret = unlikely(cond); \ 680 if (_____ret) { \ 681 if (__same_type(*(b), struct ring_buffer_per_cpu)) { \ 682 struct ring_buffer_per_cpu *__b = \ 683 (void *)b; \ 684 atomic_inc(&__b->buffer->record_disabled); \ 685 } else \ 686 atomic_inc(&b->record_disabled); \ 687 WARN_ON(1); \ 688 } \ 689 _____ret; \ 690 }) 691 692 /* Up this if you want to test the TIME_EXTENTS and normalization */ 693 #define DEBUG_SHIFT 0 694 695 static inline u64 rb_time_stamp(struct ring_buffer *buffer) 696 { 697 /* shift to debug/test normalization and TIME_EXTENTS */ 698 return buffer->clock() << DEBUG_SHIFT; 699 } 700 701 u64 ring_buffer_time_stamp(struct ring_buffer *buffer, int cpu) 702 { 703 u64 time; 704 705 preempt_disable_notrace(); 706 time = rb_time_stamp(buffer); 707 preempt_enable_no_resched_notrace(); 708 709 return time; 710 } 711 EXPORT_SYMBOL_GPL(ring_buffer_time_stamp); 712 713 void ring_buffer_normalize_time_stamp(struct ring_buffer *buffer, 714 int cpu, u64 *ts) 715 { 716 /* Just stupid testing the normalize function and deltas */ 717 *ts >>= DEBUG_SHIFT; 718 } 719 EXPORT_SYMBOL_GPL(ring_buffer_normalize_time_stamp); 720 721 /* 722 * Making the ring buffer lockless makes things tricky. 723 * Although writes only happen on the CPU that they are on, 724 * and they only need to worry about interrupts. Reads can 725 * happen on any CPU. 726 * 727 * The reader page is always off the ring buffer, but when the 728 * reader finishes with a page, it needs to swap its page with 729 * a new one from the buffer. The reader needs to take from 730 * the head (writes go to the tail). But if a writer is in overwrite 731 * mode and wraps, it must push the head page forward. 732 * 733 * Here lies the problem. 734 * 735 * The reader must be careful to replace only the head page, and 736 * not another one. As described at the top of the file in the 737 * ASCII art, the reader sets its old page to point to the next 738 * page after head. It then sets the page after head to point to 739 * the old reader page. But if the writer moves the head page 740 * during this operation, the reader could end up with the tail. 741 * 742 * We use cmpxchg to help prevent this race. We also do something 743 * special with the page before head. We set the LSB to 1. 744 * 745 * When the writer must push the page forward, it will clear the 746 * bit that points to the head page, move the head, and then set 747 * the bit that points to the new head page. 748 * 749 * We also don't want an interrupt coming in and moving the head 750 * page on another writer. Thus we use the second LSB to catch 751 * that too. Thus: 752 * 753 * head->list->prev->next bit 1 bit 0 754 * ------- ------- 755 * Normal page 0 0 756 * Points to head page 0 1 757 * New head page 1 0 758 * 759 * Note we can not trust the prev pointer of the head page, because: 760 * 761 * +----+ +-----+ +-----+ 762 * | |------>| T |---X--->| N | 763 * | |<------| | | | 764 * +----+ +-----+ +-----+ 765 * ^ ^ | 766 * | +-----+ | | 767 * +----------| R |----------+ | 768 * | |<-----------+ 769 * +-----+ 770 * 771 * Key: ---X--> HEAD flag set in pointer 772 * T Tail page 773 * R Reader page 774 * N Next page 775 * 776 * (see __rb_reserve_next() to see where this happens) 777 * 778 * What the above shows is that the reader just swapped out 779 * the reader page with a page in the buffer, but before it 780 * could make the new header point back to the new page added 781 * it was preempted by a writer. The writer moved forward onto 782 * the new page added by the reader and is about to move forward 783 * again. 784 * 785 * You can see, it is legitimate for the previous pointer of 786 * the head (or any page) not to point back to itself. But only 787 * temporarially. 788 */ 789 790 #define RB_PAGE_NORMAL 0UL 791 #define RB_PAGE_HEAD 1UL 792 #define RB_PAGE_UPDATE 2UL 793 794 795 #define RB_FLAG_MASK 3UL 796 797 /* PAGE_MOVED is not part of the mask */ 798 #define RB_PAGE_MOVED 4UL 799 800 /* 801 * rb_list_head - remove any bit 802 */ 803 static struct list_head *rb_list_head(struct list_head *list) 804 { 805 unsigned long val = (unsigned long)list; 806 807 return (struct list_head *)(val & ~RB_FLAG_MASK); 808 } 809 810 /* 811 * rb_is_head_page - test if the given page is the head page 812 * 813 * Because the reader may move the head_page pointer, we can 814 * not trust what the head page is (it may be pointing to 815 * the reader page). But if the next page is a header page, 816 * its flags will be non zero. 817 */ 818 static inline int 819 rb_is_head_page(struct ring_buffer_per_cpu *cpu_buffer, 820 struct buffer_page *page, struct list_head *list) 821 { 822 unsigned long val; 823 824 val = (unsigned long)list->next; 825 826 if ((val & ~RB_FLAG_MASK) != (unsigned long)&page->list) 827 return RB_PAGE_MOVED; 828 829 return val & RB_FLAG_MASK; 830 } 831 832 /* 833 * rb_is_reader_page 834 * 835 * The unique thing about the reader page, is that, if the 836 * writer is ever on it, the previous pointer never points 837 * back to the reader page. 838 */ 839 static int rb_is_reader_page(struct buffer_page *page) 840 { 841 struct list_head *list = page->list.prev; 842 843 return rb_list_head(list->next) != &page->list; 844 } 845 846 /* 847 * rb_set_list_to_head - set a list_head to be pointing to head. 848 */ 849 static void rb_set_list_to_head(struct ring_buffer_per_cpu *cpu_buffer, 850 struct list_head *list) 851 { 852 unsigned long *ptr; 853 854 ptr = (unsigned long *)&list->next; 855 *ptr |= RB_PAGE_HEAD; 856 *ptr &= ~RB_PAGE_UPDATE; 857 } 858 859 /* 860 * rb_head_page_activate - sets up head page 861 */ 862 static void rb_head_page_activate(struct ring_buffer_per_cpu *cpu_buffer) 863 { 864 struct buffer_page *head; 865 866 head = cpu_buffer->head_page; 867 if (!head) 868 return; 869 870 /* 871 * Set the previous list pointer to have the HEAD flag. 872 */ 873 rb_set_list_to_head(cpu_buffer, head->list.prev); 874 } 875 876 static void rb_list_head_clear(struct list_head *list) 877 { 878 unsigned long *ptr = (unsigned long *)&list->next; 879 880 *ptr &= ~RB_FLAG_MASK; 881 } 882 883 /* 884 * rb_head_page_dactivate - clears head page ptr (for free list) 885 */ 886 static void 887 rb_head_page_deactivate(struct ring_buffer_per_cpu *cpu_buffer) 888 { 889 struct list_head *hd; 890 891 /* Go through the whole list and clear any pointers found. */ 892 rb_list_head_clear(cpu_buffer->pages); 893 894 list_for_each(hd, cpu_buffer->pages) 895 rb_list_head_clear(hd); 896 } 897 898 static int rb_head_page_set(struct ring_buffer_per_cpu *cpu_buffer, 899 struct buffer_page *head, 900 struct buffer_page *prev, 901 int old_flag, int new_flag) 902 { 903 struct list_head *list; 904 unsigned long val = (unsigned long)&head->list; 905 unsigned long ret; 906 907 list = &prev->list; 908 909 val &= ~RB_FLAG_MASK; 910 911 ret = cmpxchg((unsigned long *)&list->next, 912 val | old_flag, val | new_flag); 913 914 /* check if the reader took the page */ 915 if ((ret & ~RB_FLAG_MASK) != val) 916 return RB_PAGE_MOVED; 917 918 return ret & RB_FLAG_MASK; 919 } 920 921 static int rb_head_page_set_update(struct ring_buffer_per_cpu *cpu_buffer, 922 struct buffer_page *head, 923 struct buffer_page *prev, 924 int old_flag) 925 { 926 return rb_head_page_set(cpu_buffer, head, prev, 927 old_flag, RB_PAGE_UPDATE); 928 } 929 930 static int rb_head_page_set_head(struct ring_buffer_per_cpu *cpu_buffer, 931 struct buffer_page *head, 932 struct buffer_page *prev, 933 int old_flag) 934 { 935 return rb_head_page_set(cpu_buffer, head, prev, 936 old_flag, RB_PAGE_HEAD); 937 } 938 939 static int rb_head_page_set_normal(struct ring_buffer_per_cpu *cpu_buffer, 940 struct buffer_page *head, 941 struct buffer_page *prev, 942 int old_flag) 943 { 944 return rb_head_page_set(cpu_buffer, head, prev, 945 old_flag, RB_PAGE_NORMAL); 946 } 947 948 static inline void rb_inc_page(struct ring_buffer_per_cpu *cpu_buffer, 949 struct buffer_page **bpage) 950 { 951 struct list_head *p = rb_list_head((*bpage)->list.next); 952 953 *bpage = list_entry(p, struct buffer_page, list); 954 } 955 956 static struct buffer_page * 957 rb_set_head_page(struct ring_buffer_per_cpu *cpu_buffer) 958 { 959 struct buffer_page *head; 960 struct buffer_page *page; 961 struct list_head *list; 962 int i; 963 964 if (RB_WARN_ON(cpu_buffer, !cpu_buffer->head_page)) 965 return NULL; 966 967 /* sanity check */ 968 list = cpu_buffer->pages; 969 if (RB_WARN_ON(cpu_buffer, rb_list_head(list->prev->next) != list)) 970 return NULL; 971 972 page = head = cpu_buffer->head_page; 973 /* 974 * It is possible that the writer moves the header behind 975 * where we started, and we miss in one loop. 976 * A second loop should grab the header, but we'll do 977 * three loops just because I'm paranoid. 978 */ 979 for (i = 0; i < 3; i++) { 980 do { 981 if (rb_is_head_page(cpu_buffer, page, page->list.prev)) { 982 cpu_buffer->head_page = page; 983 return page; 984 } 985 rb_inc_page(cpu_buffer, &page); 986 } while (page != head); 987 } 988 989 RB_WARN_ON(cpu_buffer, 1); 990 991 return NULL; 992 } 993 994 static int rb_head_page_replace(struct buffer_page *old, 995 struct buffer_page *new) 996 { 997 unsigned long *ptr = (unsigned long *)&old->list.prev->next; 998 unsigned long val; 999 unsigned long ret; 1000 1001 val = *ptr & ~RB_FLAG_MASK; 1002 val |= RB_PAGE_HEAD; 1003 1004 ret = cmpxchg(ptr, val, (unsigned long)&new->list); 1005 1006 return ret == val; 1007 } 1008 1009 /* 1010 * rb_tail_page_update - move the tail page forward 1011 * 1012 * Returns 1 if moved tail page, 0 if someone else did. 1013 */ 1014 static int rb_tail_page_update(struct ring_buffer_per_cpu *cpu_buffer, 1015 struct buffer_page *tail_page, 1016 struct buffer_page *next_page) 1017 { 1018 struct buffer_page *old_tail; 1019 unsigned long old_entries; 1020 unsigned long old_write; 1021 int ret = 0; 1022 1023 /* 1024 * The tail page now needs to be moved forward. 1025 * 1026 * We need to reset the tail page, but without messing 1027 * with possible erasing of data brought in by interrupts 1028 * that have moved the tail page and are currently on it. 1029 * 1030 * We add a counter to the write field to denote this. 1031 */ 1032 old_write = local_add_return(RB_WRITE_INTCNT, &next_page->write); 1033 old_entries = local_add_return(RB_WRITE_INTCNT, &next_page->entries); 1034 1035 /* 1036 * Just make sure we have seen our old_write and synchronize 1037 * with any interrupts that come in. 1038 */ 1039 barrier(); 1040 1041 /* 1042 * If the tail page is still the same as what we think 1043 * it is, then it is up to us to update the tail 1044 * pointer. 1045 */ 1046 if (tail_page == cpu_buffer->tail_page) { 1047 /* Zero the write counter */ 1048 unsigned long val = old_write & ~RB_WRITE_MASK; 1049 unsigned long eval = old_entries & ~RB_WRITE_MASK; 1050 1051 /* 1052 * This will only succeed if an interrupt did 1053 * not come in and change it. In which case, we 1054 * do not want to modify it. 1055 * 1056 * We add (void) to let the compiler know that we do not care 1057 * about the return value of these functions. We use the 1058 * cmpxchg to only update if an interrupt did not already 1059 * do it for us. If the cmpxchg fails, we don't care. 1060 */ 1061 (void)local_cmpxchg(&next_page->write, old_write, val); 1062 (void)local_cmpxchg(&next_page->entries, old_entries, eval); 1063 1064 /* 1065 * No need to worry about races with clearing out the commit. 1066 * it only can increment when a commit takes place. But that 1067 * only happens in the outer most nested commit. 1068 */ 1069 local_set(&next_page->page->commit, 0); 1070 1071 old_tail = cmpxchg(&cpu_buffer->tail_page, 1072 tail_page, next_page); 1073 1074 if (old_tail == tail_page) 1075 ret = 1; 1076 } 1077 1078 return ret; 1079 } 1080 1081 static int rb_check_bpage(struct ring_buffer_per_cpu *cpu_buffer, 1082 struct buffer_page *bpage) 1083 { 1084 unsigned long val = (unsigned long)bpage; 1085 1086 if (RB_WARN_ON(cpu_buffer, val & RB_FLAG_MASK)) 1087 return 1; 1088 1089 return 0; 1090 } 1091 1092 /** 1093 * rb_check_list - make sure a pointer to a list has the last bits zero 1094 */ 1095 static int rb_check_list(struct ring_buffer_per_cpu *cpu_buffer, 1096 struct list_head *list) 1097 { 1098 if (RB_WARN_ON(cpu_buffer, rb_list_head(list->prev) != list->prev)) 1099 return 1; 1100 if (RB_WARN_ON(cpu_buffer, rb_list_head(list->next) != list->next)) 1101 return 1; 1102 return 0; 1103 } 1104 1105 /** 1106 * rb_check_pages - integrity check of buffer pages 1107 * @cpu_buffer: CPU buffer with pages to test 1108 * 1109 * As a safety measure we check to make sure the data pages have not 1110 * been corrupted. 1111 */ 1112 static int rb_check_pages(struct ring_buffer_per_cpu *cpu_buffer) 1113 { 1114 struct list_head *head = cpu_buffer->pages; 1115 struct buffer_page *bpage, *tmp; 1116 1117 /* Reset the head page if it exists */ 1118 if (cpu_buffer->head_page) 1119 rb_set_head_page(cpu_buffer); 1120 1121 rb_head_page_deactivate(cpu_buffer); 1122 1123 if (RB_WARN_ON(cpu_buffer, head->next->prev != head)) 1124 return -1; 1125 if (RB_WARN_ON(cpu_buffer, head->prev->next != head)) 1126 return -1; 1127 1128 if (rb_check_list(cpu_buffer, head)) 1129 return -1; 1130 1131 list_for_each_entry_safe(bpage, tmp, head, list) { 1132 if (RB_WARN_ON(cpu_buffer, 1133 bpage->list.next->prev != &bpage->list)) 1134 return -1; 1135 if (RB_WARN_ON(cpu_buffer, 1136 bpage->list.prev->next != &bpage->list)) 1137 return -1; 1138 if (rb_check_list(cpu_buffer, &bpage->list)) 1139 return -1; 1140 } 1141 1142 rb_head_page_activate(cpu_buffer); 1143 1144 return 0; 1145 } 1146 1147 static int __rb_allocate_pages(int nr_pages, struct list_head *pages, int cpu) 1148 { 1149 int i; 1150 struct buffer_page *bpage, *tmp; 1151 1152 for (i = 0; i < nr_pages; i++) { 1153 struct page *page; 1154 /* 1155 * __GFP_NORETRY flag makes sure that the allocation fails 1156 * gracefully without invoking oom-killer and the system is 1157 * not destabilized. 1158 */ 1159 bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()), 1160 GFP_KERNEL | __GFP_NORETRY, 1161 cpu_to_node(cpu)); 1162 if (!bpage) 1163 goto free_pages; 1164 1165 list_add(&bpage->list, pages); 1166 1167 page = alloc_pages_node(cpu_to_node(cpu), 1168 GFP_KERNEL | __GFP_NORETRY, 0); 1169 if (!page) 1170 goto free_pages; 1171 bpage->page = page_address(page); 1172 rb_init_page(bpage->page); 1173 } 1174 1175 return 0; 1176 1177 free_pages: 1178 list_for_each_entry_safe(bpage, tmp, pages, list) { 1179 list_del_init(&bpage->list); 1180 free_buffer_page(bpage); 1181 } 1182 1183 return -ENOMEM; 1184 } 1185 1186 static int rb_allocate_pages(struct ring_buffer_per_cpu *cpu_buffer, 1187 unsigned nr_pages) 1188 { 1189 LIST_HEAD(pages); 1190 1191 WARN_ON(!nr_pages); 1192 1193 if (__rb_allocate_pages(nr_pages, &pages, cpu_buffer->cpu)) 1194 return -ENOMEM; 1195 1196 /* 1197 * The ring buffer page list is a circular list that does not 1198 * start and end with a list head. All page list items point to 1199 * other pages. 1200 */ 1201 cpu_buffer->pages = pages.next; 1202 list_del(&pages); 1203 1204 cpu_buffer->nr_pages = nr_pages; 1205 1206 rb_check_pages(cpu_buffer); 1207 1208 return 0; 1209 } 1210 1211 static struct ring_buffer_per_cpu * 1212 rb_allocate_cpu_buffer(struct ring_buffer *buffer, int nr_pages, int cpu) 1213 { 1214 struct ring_buffer_per_cpu *cpu_buffer; 1215 struct buffer_page *bpage; 1216 struct page *page; 1217 int ret; 1218 1219 cpu_buffer = kzalloc_node(ALIGN(sizeof(*cpu_buffer), cache_line_size()), 1220 GFP_KERNEL, cpu_to_node(cpu)); 1221 if (!cpu_buffer) 1222 return NULL; 1223 1224 cpu_buffer->cpu = cpu; 1225 cpu_buffer->buffer = buffer; 1226 raw_spin_lock_init(&cpu_buffer->reader_lock); 1227 lockdep_set_class(&cpu_buffer->reader_lock, buffer->reader_lock_key); 1228 cpu_buffer->lock = (arch_spinlock_t)__ARCH_SPIN_LOCK_UNLOCKED; 1229 INIT_WORK(&cpu_buffer->update_pages_work, update_pages_handler); 1230 init_completion(&cpu_buffer->update_done); 1231 init_irq_work(&cpu_buffer->irq_work.work, rb_wake_up_waiters); 1232 init_waitqueue_head(&cpu_buffer->irq_work.waiters); 1233 1234 bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()), 1235 GFP_KERNEL, cpu_to_node(cpu)); 1236 if (!bpage) 1237 goto fail_free_buffer; 1238 1239 rb_check_bpage(cpu_buffer, bpage); 1240 1241 cpu_buffer->reader_page = bpage; 1242 page = alloc_pages_node(cpu_to_node(cpu), GFP_KERNEL, 0); 1243 if (!page) 1244 goto fail_free_reader; 1245 bpage->page = page_address(page); 1246 rb_init_page(bpage->page); 1247 1248 INIT_LIST_HEAD(&cpu_buffer->reader_page->list); 1249 INIT_LIST_HEAD(&cpu_buffer->new_pages); 1250 1251 ret = rb_allocate_pages(cpu_buffer, nr_pages); 1252 if (ret < 0) 1253 goto fail_free_reader; 1254 1255 cpu_buffer->head_page 1256 = list_entry(cpu_buffer->pages, struct buffer_page, list); 1257 cpu_buffer->tail_page = cpu_buffer->commit_page = cpu_buffer->head_page; 1258 1259 rb_head_page_activate(cpu_buffer); 1260 1261 return cpu_buffer; 1262 1263 fail_free_reader: 1264 free_buffer_page(cpu_buffer->reader_page); 1265 1266 fail_free_buffer: 1267 kfree(cpu_buffer); 1268 return NULL; 1269 } 1270 1271 static void rb_free_cpu_buffer(struct ring_buffer_per_cpu *cpu_buffer) 1272 { 1273 struct list_head *head = cpu_buffer->pages; 1274 struct buffer_page *bpage, *tmp; 1275 1276 free_buffer_page(cpu_buffer->reader_page); 1277 1278 rb_head_page_deactivate(cpu_buffer); 1279 1280 if (head) { 1281 list_for_each_entry_safe(bpage, tmp, head, list) { 1282 list_del_init(&bpage->list); 1283 free_buffer_page(bpage); 1284 } 1285 bpage = list_entry(head, struct buffer_page, list); 1286 free_buffer_page(bpage); 1287 } 1288 1289 kfree(cpu_buffer); 1290 } 1291 1292 #ifdef CONFIG_HOTPLUG_CPU 1293 static int rb_cpu_notify(struct notifier_block *self, 1294 unsigned long action, void *hcpu); 1295 #endif 1296 1297 /** 1298 * __ring_buffer_alloc - allocate a new ring_buffer 1299 * @size: the size in bytes per cpu that is needed. 1300 * @flags: attributes to set for the ring buffer. 1301 * 1302 * Currently the only flag that is available is the RB_FL_OVERWRITE 1303 * flag. This flag means that the buffer will overwrite old data 1304 * when the buffer wraps. If this flag is not set, the buffer will 1305 * drop data when the tail hits the head. 1306 */ 1307 struct ring_buffer *__ring_buffer_alloc(unsigned long size, unsigned flags, 1308 struct lock_class_key *key) 1309 { 1310 struct ring_buffer *buffer; 1311 int bsize; 1312 int cpu, nr_pages; 1313 1314 /* keep it in its own cache line */ 1315 buffer = kzalloc(ALIGN(sizeof(*buffer), cache_line_size()), 1316 GFP_KERNEL); 1317 if (!buffer) 1318 return NULL; 1319 1320 if (!alloc_cpumask_var(&buffer->cpumask, GFP_KERNEL)) 1321 goto fail_free_buffer; 1322 1323 nr_pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE); 1324 buffer->flags = flags; 1325 buffer->clock = trace_clock_local; 1326 buffer->reader_lock_key = key; 1327 1328 init_irq_work(&buffer->irq_work.work, rb_wake_up_waiters); 1329 init_waitqueue_head(&buffer->irq_work.waiters); 1330 1331 /* need at least two pages */ 1332 if (nr_pages < 2) 1333 nr_pages = 2; 1334 1335 /* 1336 * In case of non-hotplug cpu, if the ring-buffer is allocated 1337 * in early initcall, it will not be notified of secondary cpus. 1338 * In that off case, we need to allocate for all possible cpus. 1339 */ 1340 #ifdef CONFIG_HOTPLUG_CPU 1341 cpu_notifier_register_begin(); 1342 cpumask_copy(buffer->cpumask, cpu_online_mask); 1343 #else 1344 cpumask_copy(buffer->cpumask, cpu_possible_mask); 1345 #endif 1346 buffer->cpus = nr_cpu_ids; 1347 1348 bsize = sizeof(void *) * nr_cpu_ids; 1349 buffer->buffers = kzalloc(ALIGN(bsize, cache_line_size()), 1350 GFP_KERNEL); 1351 if (!buffer->buffers) 1352 goto fail_free_cpumask; 1353 1354 for_each_buffer_cpu(buffer, cpu) { 1355 buffer->buffers[cpu] = 1356 rb_allocate_cpu_buffer(buffer, nr_pages, cpu); 1357 if (!buffer->buffers[cpu]) 1358 goto fail_free_buffers; 1359 } 1360 1361 #ifdef CONFIG_HOTPLUG_CPU 1362 buffer->cpu_notify.notifier_call = rb_cpu_notify; 1363 buffer->cpu_notify.priority = 0; 1364 __register_cpu_notifier(&buffer->cpu_notify); 1365 cpu_notifier_register_done(); 1366 #endif 1367 1368 mutex_init(&buffer->mutex); 1369 1370 return buffer; 1371 1372 fail_free_buffers: 1373 for_each_buffer_cpu(buffer, cpu) { 1374 if (buffer->buffers[cpu]) 1375 rb_free_cpu_buffer(buffer->buffers[cpu]); 1376 } 1377 kfree(buffer->buffers); 1378 1379 fail_free_cpumask: 1380 free_cpumask_var(buffer->cpumask); 1381 #ifdef CONFIG_HOTPLUG_CPU 1382 cpu_notifier_register_done(); 1383 #endif 1384 1385 fail_free_buffer: 1386 kfree(buffer); 1387 return NULL; 1388 } 1389 EXPORT_SYMBOL_GPL(__ring_buffer_alloc); 1390 1391 /** 1392 * ring_buffer_free - free a ring buffer. 1393 * @buffer: the buffer to free. 1394 */ 1395 void 1396 ring_buffer_free(struct ring_buffer *buffer) 1397 { 1398 int cpu; 1399 1400 #ifdef CONFIG_HOTPLUG_CPU 1401 cpu_notifier_register_begin(); 1402 __unregister_cpu_notifier(&buffer->cpu_notify); 1403 #endif 1404 1405 for_each_buffer_cpu(buffer, cpu) 1406 rb_free_cpu_buffer(buffer->buffers[cpu]); 1407 1408 #ifdef CONFIG_HOTPLUG_CPU 1409 cpu_notifier_register_done(); 1410 #endif 1411 1412 kfree(buffer->buffers); 1413 free_cpumask_var(buffer->cpumask); 1414 1415 kfree(buffer); 1416 } 1417 EXPORT_SYMBOL_GPL(ring_buffer_free); 1418 1419 void ring_buffer_set_clock(struct ring_buffer *buffer, 1420 u64 (*clock)(void)) 1421 { 1422 buffer->clock = clock; 1423 } 1424 1425 static void rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer); 1426 1427 static inline unsigned long rb_page_entries(struct buffer_page *bpage) 1428 { 1429 return local_read(&bpage->entries) & RB_WRITE_MASK; 1430 } 1431 1432 static inline unsigned long rb_page_write(struct buffer_page *bpage) 1433 { 1434 return local_read(&bpage->write) & RB_WRITE_MASK; 1435 } 1436 1437 static int 1438 rb_remove_pages(struct ring_buffer_per_cpu *cpu_buffer, unsigned int nr_pages) 1439 { 1440 struct list_head *tail_page, *to_remove, *next_page; 1441 struct buffer_page *to_remove_page, *tmp_iter_page; 1442 struct buffer_page *last_page, *first_page; 1443 unsigned int nr_removed; 1444 unsigned long head_bit; 1445 int page_entries; 1446 1447 head_bit = 0; 1448 1449 raw_spin_lock_irq(&cpu_buffer->reader_lock); 1450 atomic_inc(&cpu_buffer->record_disabled); 1451 /* 1452 * We don't race with the readers since we have acquired the reader 1453 * lock. We also don't race with writers after disabling recording. 1454 * This makes it easy to figure out the first and the last page to be 1455 * removed from the list. We unlink all the pages in between including 1456 * the first and last pages. This is done in a busy loop so that we 1457 * lose the least number of traces. 1458 * The pages are freed after we restart recording and unlock readers. 1459 */ 1460 tail_page = &cpu_buffer->tail_page->list; 1461 1462 /* 1463 * tail page might be on reader page, we remove the next page 1464 * from the ring buffer 1465 */ 1466 if (cpu_buffer->tail_page == cpu_buffer->reader_page) 1467 tail_page = rb_list_head(tail_page->next); 1468 to_remove = tail_page; 1469 1470 /* start of pages to remove */ 1471 first_page = list_entry(rb_list_head(to_remove->next), 1472 struct buffer_page, list); 1473 1474 for (nr_removed = 0; nr_removed < nr_pages; nr_removed++) { 1475 to_remove = rb_list_head(to_remove)->next; 1476 head_bit |= (unsigned long)to_remove & RB_PAGE_HEAD; 1477 } 1478 1479 next_page = rb_list_head(to_remove)->next; 1480 1481 /* 1482 * Now we remove all pages between tail_page and next_page. 1483 * Make sure that we have head_bit value preserved for the 1484 * next page 1485 */ 1486 tail_page->next = (struct list_head *)((unsigned long)next_page | 1487 head_bit); 1488 next_page = rb_list_head(next_page); 1489 next_page->prev = tail_page; 1490 1491 /* make sure pages points to a valid page in the ring buffer */ 1492 cpu_buffer->pages = next_page; 1493 1494 /* update head page */ 1495 if (head_bit) 1496 cpu_buffer->head_page = list_entry(next_page, 1497 struct buffer_page, list); 1498 1499 /* 1500 * change read pointer to make sure any read iterators reset 1501 * themselves 1502 */ 1503 cpu_buffer->read = 0; 1504 1505 /* pages are removed, resume tracing and then free the pages */ 1506 atomic_dec(&cpu_buffer->record_disabled); 1507 raw_spin_unlock_irq(&cpu_buffer->reader_lock); 1508 1509 RB_WARN_ON(cpu_buffer, list_empty(cpu_buffer->pages)); 1510 1511 /* last buffer page to remove */ 1512 last_page = list_entry(rb_list_head(to_remove), struct buffer_page, 1513 list); 1514 tmp_iter_page = first_page; 1515 1516 do { 1517 to_remove_page = tmp_iter_page; 1518 rb_inc_page(cpu_buffer, &tmp_iter_page); 1519 1520 /* update the counters */ 1521 page_entries = rb_page_entries(to_remove_page); 1522 if (page_entries) { 1523 /* 1524 * If something was added to this page, it was full 1525 * since it is not the tail page. So we deduct the 1526 * bytes consumed in ring buffer from here. 1527 * Increment overrun to account for the lost events. 1528 */ 1529 local_add(page_entries, &cpu_buffer->overrun); 1530 local_sub(BUF_PAGE_SIZE, &cpu_buffer->entries_bytes); 1531 } 1532 1533 /* 1534 * We have already removed references to this list item, just 1535 * free up the buffer_page and its page 1536 */ 1537 free_buffer_page(to_remove_page); 1538 nr_removed--; 1539 1540 } while (to_remove_page != last_page); 1541 1542 RB_WARN_ON(cpu_buffer, nr_removed); 1543 1544 return nr_removed == 0; 1545 } 1546 1547 static int 1548 rb_insert_pages(struct ring_buffer_per_cpu *cpu_buffer) 1549 { 1550 struct list_head *pages = &cpu_buffer->new_pages; 1551 int retries, success; 1552 1553 raw_spin_lock_irq(&cpu_buffer->reader_lock); 1554 /* 1555 * We are holding the reader lock, so the reader page won't be swapped 1556 * in the ring buffer. Now we are racing with the writer trying to 1557 * move head page and the tail page. 1558 * We are going to adapt the reader page update process where: 1559 * 1. We first splice the start and end of list of new pages between 1560 * the head page and its previous page. 1561 * 2. We cmpxchg the prev_page->next to point from head page to the 1562 * start of new pages list. 1563 * 3. Finally, we update the head->prev to the end of new list. 1564 * 1565 * We will try this process 10 times, to make sure that we don't keep 1566 * spinning. 1567 */ 1568 retries = 10; 1569 success = 0; 1570 while (retries--) { 1571 struct list_head *head_page, *prev_page, *r; 1572 struct list_head *last_page, *first_page; 1573 struct list_head *head_page_with_bit; 1574 1575 head_page = &rb_set_head_page(cpu_buffer)->list; 1576 if (!head_page) 1577 break; 1578 prev_page = head_page->prev; 1579 1580 first_page = pages->next; 1581 last_page = pages->prev; 1582 1583 head_page_with_bit = (struct list_head *) 1584 ((unsigned long)head_page | RB_PAGE_HEAD); 1585 1586 last_page->next = head_page_with_bit; 1587 first_page->prev = prev_page; 1588 1589 r = cmpxchg(&prev_page->next, head_page_with_bit, first_page); 1590 1591 if (r == head_page_with_bit) { 1592 /* 1593 * yay, we replaced the page pointer to our new list, 1594 * now, we just have to update to head page's prev 1595 * pointer to point to end of list 1596 */ 1597 head_page->prev = last_page; 1598 success = 1; 1599 break; 1600 } 1601 } 1602 1603 if (success) 1604 INIT_LIST_HEAD(pages); 1605 /* 1606 * If we weren't successful in adding in new pages, warn and stop 1607 * tracing 1608 */ 1609 RB_WARN_ON(cpu_buffer, !success); 1610 raw_spin_unlock_irq(&cpu_buffer->reader_lock); 1611 1612 /* free pages if they weren't inserted */ 1613 if (!success) { 1614 struct buffer_page *bpage, *tmp; 1615 list_for_each_entry_safe(bpage, tmp, &cpu_buffer->new_pages, 1616 list) { 1617 list_del_init(&bpage->list); 1618 free_buffer_page(bpage); 1619 } 1620 } 1621 return success; 1622 } 1623 1624 static void rb_update_pages(struct ring_buffer_per_cpu *cpu_buffer) 1625 { 1626 int success; 1627 1628 if (cpu_buffer->nr_pages_to_update > 0) 1629 success = rb_insert_pages(cpu_buffer); 1630 else 1631 success = rb_remove_pages(cpu_buffer, 1632 -cpu_buffer->nr_pages_to_update); 1633 1634 if (success) 1635 cpu_buffer->nr_pages += cpu_buffer->nr_pages_to_update; 1636 } 1637 1638 static void update_pages_handler(struct work_struct *work) 1639 { 1640 struct ring_buffer_per_cpu *cpu_buffer = container_of(work, 1641 struct ring_buffer_per_cpu, update_pages_work); 1642 rb_update_pages(cpu_buffer); 1643 complete(&cpu_buffer->update_done); 1644 } 1645 1646 /** 1647 * ring_buffer_resize - resize the ring buffer 1648 * @buffer: the buffer to resize. 1649 * @size: the new size. 1650 * @cpu_id: the cpu buffer to resize 1651 * 1652 * Minimum size is 2 * BUF_PAGE_SIZE. 1653 * 1654 * Returns 0 on success and < 0 on failure. 1655 */ 1656 int ring_buffer_resize(struct ring_buffer *buffer, unsigned long size, 1657 int cpu_id) 1658 { 1659 struct ring_buffer_per_cpu *cpu_buffer; 1660 unsigned nr_pages; 1661 int cpu, err = 0; 1662 1663 /* 1664 * Always succeed at resizing a non-existent buffer: 1665 */ 1666 if (!buffer) 1667 return size; 1668 1669 /* Make sure the requested buffer exists */ 1670 if (cpu_id != RING_BUFFER_ALL_CPUS && 1671 !cpumask_test_cpu(cpu_id, buffer->cpumask)) 1672 return size; 1673 1674 size = DIV_ROUND_UP(size, BUF_PAGE_SIZE); 1675 size *= BUF_PAGE_SIZE; 1676 1677 /* we need a minimum of two pages */ 1678 if (size < BUF_PAGE_SIZE * 2) 1679 size = BUF_PAGE_SIZE * 2; 1680 1681 nr_pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE); 1682 1683 /* 1684 * Don't succeed if resizing is disabled, as a reader might be 1685 * manipulating the ring buffer and is expecting a sane state while 1686 * this is true. 1687 */ 1688 if (atomic_read(&buffer->resize_disabled)) 1689 return -EBUSY; 1690 1691 /* prevent another thread from changing buffer sizes */ 1692 mutex_lock(&buffer->mutex); 1693 1694 if (cpu_id == RING_BUFFER_ALL_CPUS) { 1695 /* calculate the pages to update */ 1696 for_each_buffer_cpu(buffer, cpu) { 1697 cpu_buffer = buffer->buffers[cpu]; 1698 1699 cpu_buffer->nr_pages_to_update = nr_pages - 1700 cpu_buffer->nr_pages; 1701 /* 1702 * nothing more to do for removing pages or no update 1703 */ 1704 if (cpu_buffer->nr_pages_to_update <= 0) 1705 continue; 1706 /* 1707 * to add pages, make sure all new pages can be 1708 * allocated without receiving ENOMEM 1709 */ 1710 INIT_LIST_HEAD(&cpu_buffer->new_pages); 1711 if (__rb_allocate_pages(cpu_buffer->nr_pages_to_update, 1712 &cpu_buffer->new_pages, cpu)) { 1713 /* not enough memory for new pages */ 1714 err = -ENOMEM; 1715 goto out_err; 1716 } 1717 } 1718 1719 get_online_cpus(); 1720 /* 1721 * Fire off all the required work handlers 1722 * We can't schedule on offline CPUs, but it's not necessary 1723 * since we can change their buffer sizes without any race. 1724 */ 1725 for_each_buffer_cpu(buffer, cpu) { 1726 cpu_buffer = buffer->buffers[cpu]; 1727 if (!cpu_buffer->nr_pages_to_update) 1728 continue; 1729 1730 /* Can't run something on an offline CPU. */ 1731 if (!cpu_online(cpu)) { 1732 rb_update_pages(cpu_buffer); 1733 cpu_buffer->nr_pages_to_update = 0; 1734 } else { 1735 schedule_work_on(cpu, 1736 &cpu_buffer->update_pages_work); 1737 } 1738 } 1739 1740 /* wait for all the updates to complete */ 1741 for_each_buffer_cpu(buffer, cpu) { 1742 cpu_buffer = buffer->buffers[cpu]; 1743 if (!cpu_buffer->nr_pages_to_update) 1744 continue; 1745 1746 if (cpu_online(cpu)) 1747 wait_for_completion(&cpu_buffer->update_done); 1748 cpu_buffer->nr_pages_to_update = 0; 1749 } 1750 1751 put_online_cpus(); 1752 } else { 1753 /* Make sure this CPU has been intitialized */ 1754 if (!cpumask_test_cpu(cpu_id, buffer->cpumask)) 1755 goto out; 1756 1757 cpu_buffer = buffer->buffers[cpu_id]; 1758 1759 if (nr_pages == cpu_buffer->nr_pages) 1760 goto out; 1761 1762 cpu_buffer->nr_pages_to_update = nr_pages - 1763 cpu_buffer->nr_pages; 1764 1765 INIT_LIST_HEAD(&cpu_buffer->new_pages); 1766 if (cpu_buffer->nr_pages_to_update > 0 && 1767 __rb_allocate_pages(cpu_buffer->nr_pages_to_update, 1768 &cpu_buffer->new_pages, cpu_id)) { 1769 err = -ENOMEM; 1770 goto out_err; 1771 } 1772 1773 get_online_cpus(); 1774 1775 /* Can't run something on an offline CPU. */ 1776 if (!cpu_online(cpu_id)) 1777 rb_update_pages(cpu_buffer); 1778 else { 1779 schedule_work_on(cpu_id, 1780 &cpu_buffer->update_pages_work); 1781 wait_for_completion(&cpu_buffer->update_done); 1782 } 1783 1784 cpu_buffer->nr_pages_to_update = 0; 1785 put_online_cpus(); 1786 } 1787 1788 out: 1789 /* 1790 * The ring buffer resize can happen with the ring buffer 1791 * enabled, so that the update disturbs the tracing as little 1792 * as possible. But if the buffer is disabled, we do not need 1793 * to worry about that, and we can take the time to verify 1794 * that the buffer is not corrupt. 1795 */ 1796 if (atomic_read(&buffer->record_disabled)) { 1797 atomic_inc(&buffer->record_disabled); 1798 /* 1799 * Even though the buffer was disabled, we must make sure 1800 * that it is truly disabled before calling rb_check_pages. 1801 * There could have been a race between checking 1802 * record_disable and incrementing it. 1803 */ 1804 synchronize_sched(); 1805 for_each_buffer_cpu(buffer, cpu) { 1806 cpu_buffer = buffer->buffers[cpu]; 1807 rb_check_pages(cpu_buffer); 1808 } 1809 atomic_dec(&buffer->record_disabled); 1810 } 1811 1812 mutex_unlock(&buffer->mutex); 1813 return size; 1814 1815 out_err: 1816 for_each_buffer_cpu(buffer, cpu) { 1817 struct buffer_page *bpage, *tmp; 1818 1819 cpu_buffer = buffer->buffers[cpu]; 1820 cpu_buffer->nr_pages_to_update = 0; 1821 1822 if (list_empty(&cpu_buffer->new_pages)) 1823 continue; 1824 1825 list_for_each_entry_safe(bpage, tmp, &cpu_buffer->new_pages, 1826 list) { 1827 list_del_init(&bpage->list); 1828 free_buffer_page(bpage); 1829 } 1830 } 1831 mutex_unlock(&buffer->mutex); 1832 return err; 1833 } 1834 EXPORT_SYMBOL_GPL(ring_buffer_resize); 1835 1836 void ring_buffer_change_overwrite(struct ring_buffer *buffer, int val) 1837 { 1838 mutex_lock(&buffer->mutex); 1839 if (val) 1840 buffer->flags |= RB_FL_OVERWRITE; 1841 else 1842 buffer->flags &= ~RB_FL_OVERWRITE; 1843 mutex_unlock(&buffer->mutex); 1844 } 1845 EXPORT_SYMBOL_GPL(ring_buffer_change_overwrite); 1846 1847 static inline void * 1848 __rb_data_page_index(struct buffer_data_page *bpage, unsigned index) 1849 { 1850 return bpage->data + index; 1851 } 1852 1853 static inline void *__rb_page_index(struct buffer_page *bpage, unsigned index) 1854 { 1855 return bpage->page->data + index; 1856 } 1857 1858 static inline struct ring_buffer_event * 1859 rb_reader_event(struct ring_buffer_per_cpu *cpu_buffer) 1860 { 1861 return __rb_page_index(cpu_buffer->reader_page, 1862 cpu_buffer->reader_page->read); 1863 } 1864 1865 static inline struct ring_buffer_event * 1866 rb_iter_head_event(struct ring_buffer_iter *iter) 1867 { 1868 return __rb_page_index(iter->head_page, iter->head); 1869 } 1870 1871 static inline unsigned rb_page_commit(struct buffer_page *bpage) 1872 { 1873 return local_read(&bpage->page->commit); 1874 } 1875 1876 /* Size is determined by what has been committed */ 1877 static inline unsigned rb_page_size(struct buffer_page *bpage) 1878 { 1879 return rb_page_commit(bpage); 1880 } 1881 1882 static inline unsigned 1883 rb_commit_index(struct ring_buffer_per_cpu *cpu_buffer) 1884 { 1885 return rb_page_commit(cpu_buffer->commit_page); 1886 } 1887 1888 static inline unsigned 1889 rb_event_index(struct ring_buffer_event *event) 1890 { 1891 unsigned long addr = (unsigned long)event; 1892 1893 return (addr & ~PAGE_MASK) - BUF_PAGE_HDR_SIZE; 1894 } 1895 1896 static inline int 1897 rb_event_is_commit(struct ring_buffer_per_cpu *cpu_buffer, 1898 struct ring_buffer_event *event) 1899 { 1900 unsigned long addr = (unsigned long)event; 1901 unsigned long index; 1902 1903 index = rb_event_index(event); 1904 addr &= PAGE_MASK; 1905 1906 return cpu_buffer->commit_page->page == (void *)addr && 1907 rb_commit_index(cpu_buffer) == index; 1908 } 1909 1910 static void 1911 rb_set_commit_to_write(struct ring_buffer_per_cpu *cpu_buffer) 1912 { 1913 unsigned long max_count; 1914 1915 /* 1916 * We only race with interrupts and NMIs on this CPU. 1917 * If we own the commit event, then we can commit 1918 * all others that interrupted us, since the interruptions 1919 * are in stack format (they finish before they come 1920 * back to us). This allows us to do a simple loop to 1921 * assign the commit to the tail. 1922 */ 1923 again: 1924 max_count = cpu_buffer->nr_pages * 100; 1925 1926 while (cpu_buffer->commit_page != cpu_buffer->tail_page) { 1927 if (RB_WARN_ON(cpu_buffer, !(--max_count))) 1928 return; 1929 if (RB_WARN_ON(cpu_buffer, 1930 rb_is_reader_page(cpu_buffer->tail_page))) 1931 return; 1932 local_set(&cpu_buffer->commit_page->page->commit, 1933 rb_page_write(cpu_buffer->commit_page)); 1934 rb_inc_page(cpu_buffer, &cpu_buffer->commit_page); 1935 cpu_buffer->write_stamp = 1936 cpu_buffer->commit_page->page->time_stamp; 1937 /* add barrier to keep gcc from optimizing too much */ 1938 barrier(); 1939 } 1940 while (rb_commit_index(cpu_buffer) != 1941 rb_page_write(cpu_buffer->commit_page)) { 1942 1943 local_set(&cpu_buffer->commit_page->page->commit, 1944 rb_page_write(cpu_buffer->commit_page)); 1945 RB_WARN_ON(cpu_buffer, 1946 local_read(&cpu_buffer->commit_page->page->commit) & 1947 ~RB_WRITE_MASK); 1948 barrier(); 1949 } 1950 1951 /* again, keep gcc from optimizing */ 1952 barrier(); 1953 1954 /* 1955 * If an interrupt came in just after the first while loop 1956 * and pushed the tail page forward, we will be left with 1957 * a dangling commit that will never go forward. 1958 */ 1959 if (unlikely(cpu_buffer->commit_page != cpu_buffer->tail_page)) 1960 goto again; 1961 } 1962 1963 static void rb_reset_reader_page(struct ring_buffer_per_cpu *cpu_buffer) 1964 { 1965 cpu_buffer->read_stamp = cpu_buffer->reader_page->page->time_stamp; 1966 cpu_buffer->reader_page->read = 0; 1967 } 1968 1969 static void rb_inc_iter(struct ring_buffer_iter *iter) 1970 { 1971 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer; 1972 1973 /* 1974 * The iterator could be on the reader page (it starts there). 1975 * But the head could have moved, since the reader was 1976 * found. Check for this case and assign the iterator 1977 * to the head page instead of next. 1978 */ 1979 if (iter->head_page == cpu_buffer->reader_page) 1980 iter->head_page = rb_set_head_page(cpu_buffer); 1981 else 1982 rb_inc_page(cpu_buffer, &iter->head_page); 1983 1984 iter->read_stamp = iter->head_page->page->time_stamp; 1985 iter->head = 0; 1986 } 1987 1988 /* Slow path, do not inline */ 1989 static noinline struct ring_buffer_event * 1990 rb_add_time_stamp(struct ring_buffer_event *event, u64 delta) 1991 { 1992 event->type_len = RINGBUF_TYPE_TIME_EXTEND; 1993 1994 /* Not the first event on the page? */ 1995 if (rb_event_index(event)) { 1996 event->time_delta = delta & TS_MASK; 1997 event->array[0] = delta >> TS_SHIFT; 1998 } else { 1999 /* nope, just zero it */ 2000 event->time_delta = 0; 2001 event->array[0] = 0; 2002 } 2003 2004 return skip_time_extend(event); 2005 } 2006 2007 /** 2008 * rb_update_event - update event type and data 2009 * @event: the event to update 2010 * @type: the type of event 2011 * @length: the size of the event field in the ring buffer 2012 * 2013 * Update the type and data fields of the event. The length 2014 * is the actual size that is written to the ring buffer, 2015 * and with this, we can determine what to place into the 2016 * data field. 2017 */ 2018 static void 2019 rb_update_event(struct ring_buffer_per_cpu *cpu_buffer, 2020 struct ring_buffer_event *event, unsigned length, 2021 int add_timestamp, u64 delta) 2022 { 2023 /* Only a commit updates the timestamp */ 2024 if (unlikely(!rb_event_is_commit(cpu_buffer, event))) 2025 delta = 0; 2026 2027 /* 2028 * If we need to add a timestamp, then we 2029 * add it to the start of the resevered space. 2030 */ 2031 if (unlikely(add_timestamp)) { 2032 event = rb_add_time_stamp(event, delta); 2033 length -= RB_LEN_TIME_EXTEND; 2034 delta = 0; 2035 } 2036 2037 event->time_delta = delta; 2038 length -= RB_EVNT_HDR_SIZE; 2039 if (length > RB_MAX_SMALL_DATA || RB_FORCE_8BYTE_ALIGNMENT) { 2040 event->type_len = 0; 2041 event->array[0] = length; 2042 } else 2043 event->type_len = DIV_ROUND_UP(length, RB_ALIGNMENT); 2044 } 2045 2046 /* 2047 * rb_handle_head_page - writer hit the head page 2048 * 2049 * Returns: +1 to retry page 2050 * 0 to continue 2051 * -1 on error 2052 */ 2053 static int 2054 rb_handle_head_page(struct ring_buffer_per_cpu *cpu_buffer, 2055 struct buffer_page *tail_page, 2056 struct buffer_page *next_page) 2057 { 2058 struct buffer_page *new_head; 2059 int entries; 2060 int type; 2061 int ret; 2062 2063 entries = rb_page_entries(next_page); 2064 2065 /* 2066 * The hard part is here. We need to move the head 2067 * forward, and protect against both readers on 2068 * other CPUs and writers coming in via interrupts. 2069 */ 2070 type = rb_head_page_set_update(cpu_buffer, next_page, tail_page, 2071 RB_PAGE_HEAD); 2072 2073 /* 2074 * type can be one of four: 2075 * NORMAL - an interrupt already moved it for us 2076 * HEAD - we are the first to get here. 2077 * UPDATE - we are the interrupt interrupting 2078 * a current move. 2079 * MOVED - a reader on another CPU moved the next 2080 * pointer to its reader page. Give up 2081 * and try again. 2082 */ 2083 2084 switch (type) { 2085 case RB_PAGE_HEAD: 2086 /* 2087 * We changed the head to UPDATE, thus 2088 * it is our responsibility to update 2089 * the counters. 2090 */ 2091 local_add(entries, &cpu_buffer->overrun); 2092 local_sub(BUF_PAGE_SIZE, &cpu_buffer->entries_bytes); 2093 2094 /* 2095 * The entries will be zeroed out when we move the 2096 * tail page. 2097 */ 2098 2099 /* still more to do */ 2100 break; 2101 2102 case RB_PAGE_UPDATE: 2103 /* 2104 * This is an interrupt that interrupt the 2105 * previous update. Still more to do. 2106 */ 2107 break; 2108 case RB_PAGE_NORMAL: 2109 /* 2110 * An interrupt came in before the update 2111 * and processed this for us. 2112 * Nothing left to do. 2113 */ 2114 return 1; 2115 case RB_PAGE_MOVED: 2116 /* 2117 * The reader is on another CPU and just did 2118 * a swap with our next_page. 2119 * Try again. 2120 */ 2121 return 1; 2122 default: 2123 RB_WARN_ON(cpu_buffer, 1); /* WTF??? */ 2124 return -1; 2125 } 2126 2127 /* 2128 * Now that we are here, the old head pointer is 2129 * set to UPDATE. This will keep the reader from 2130 * swapping the head page with the reader page. 2131 * The reader (on another CPU) will spin till 2132 * we are finished. 2133 * 2134 * We just need to protect against interrupts 2135 * doing the job. We will set the next pointer 2136 * to HEAD. After that, we set the old pointer 2137 * to NORMAL, but only if it was HEAD before. 2138 * otherwise we are an interrupt, and only 2139 * want the outer most commit to reset it. 2140 */ 2141 new_head = next_page; 2142 rb_inc_page(cpu_buffer, &new_head); 2143 2144 ret = rb_head_page_set_head(cpu_buffer, new_head, next_page, 2145 RB_PAGE_NORMAL); 2146 2147 /* 2148 * Valid returns are: 2149 * HEAD - an interrupt came in and already set it. 2150 * NORMAL - One of two things: 2151 * 1) We really set it. 2152 * 2) A bunch of interrupts came in and moved 2153 * the page forward again. 2154 */ 2155 switch (ret) { 2156 case RB_PAGE_HEAD: 2157 case RB_PAGE_NORMAL: 2158 /* OK */ 2159 break; 2160 default: 2161 RB_WARN_ON(cpu_buffer, 1); 2162 return -1; 2163 } 2164 2165 /* 2166 * It is possible that an interrupt came in, 2167 * set the head up, then more interrupts came in 2168 * and moved it again. When we get back here, 2169 * the page would have been set to NORMAL but we 2170 * just set it back to HEAD. 2171 * 2172 * How do you detect this? Well, if that happened 2173 * the tail page would have moved. 2174 */ 2175 if (ret == RB_PAGE_NORMAL) { 2176 /* 2177 * If the tail had moved passed next, then we need 2178 * to reset the pointer. 2179 */ 2180 if (cpu_buffer->tail_page != tail_page && 2181 cpu_buffer->tail_page != next_page) 2182 rb_head_page_set_normal(cpu_buffer, new_head, 2183 next_page, 2184 RB_PAGE_HEAD); 2185 } 2186 2187 /* 2188 * If this was the outer most commit (the one that 2189 * changed the original pointer from HEAD to UPDATE), 2190 * then it is up to us to reset it to NORMAL. 2191 */ 2192 if (type == RB_PAGE_HEAD) { 2193 ret = rb_head_page_set_normal(cpu_buffer, next_page, 2194 tail_page, 2195 RB_PAGE_UPDATE); 2196 if (RB_WARN_ON(cpu_buffer, 2197 ret != RB_PAGE_UPDATE)) 2198 return -1; 2199 } 2200 2201 return 0; 2202 } 2203 2204 static unsigned rb_calculate_event_length(unsigned length) 2205 { 2206 struct ring_buffer_event event; /* Used only for sizeof array */ 2207 2208 /* zero length can cause confusions */ 2209 if (!length) 2210 length = 1; 2211 2212 if (length > RB_MAX_SMALL_DATA || RB_FORCE_8BYTE_ALIGNMENT) 2213 length += sizeof(event.array[0]); 2214 2215 length += RB_EVNT_HDR_SIZE; 2216 length = ALIGN(length, RB_ARCH_ALIGNMENT); 2217 2218 return length; 2219 } 2220 2221 static inline void 2222 rb_reset_tail(struct ring_buffer_per_cpu *cpu_buffer, 2223 struct buffer_page *tail_page, 2224 unsigned long tail, unsigned long length) 2225 { 2226 struct ring_buffer_event *event; 2227 2228 /* 2229 * Only the event that crossed the page boundary 2230 * must fill the old tail_page with padding. 2231 */ 2232 if (tail >= BUF_PAGE_SIZE) { 2233 /* 2234 * If the page was filled, then we still need 2235 * to update the real_end. Reset it to zero 2236 * and the reader will ignore it. 2237 */ 2238 if (tail == BUF_PAGE_SIZE) 2239 tail_page->real_end = 0; 2240 2241 local_sub(length, &tail_page->write); 2242 return; 2243 } 2244 2245 event = __rb_page_index(tail_page, tail); 2246 kmemcheck_annotate_bitfield(event, bitfield); 2247 2248 /* account for padding bytes */ 2249 local_add(BUF_PAGE_SIZE - tail, &cpu_buffer->entries_bytes); 2250 2251 /* 2252 * Save the original length to the meta data. 2253 * This will be used by the reader to add lost event 2254 * counter. 2255 */ 2256 tail_page->real_end = tail; 2257 2258 /* 2259 * If this event is bigger than the minimum size, then 2260 * we need to be careful that we don't subtract the 2261 * write counter enough to allow another writer to slip 2262 * in on this page. 2263 * We put in a discarded commit instead, to make sure 2264 * that this space is not used again. 2265 * 2266 * If we are less than the minimum size, we don't need to 2267 * worry about it. 2268 */ 2269 if (tail > (BUF_PAGE_SIZE - RB_EVNT_MIN_SIZE)) { 2270 /* No room for any events */ 2271 2272 /* Mark the rest of the page with padding */ 2273 rb_event_set_padding(event); 2274 2275 /* Set the write back to the previous setting */ 2276 local_sub(length, &tail_page->write); 2277 return; 2278 } 2279 2280 /* Put in a discarded event */ 2281 event->array[0] = (BUF_PAGE_SIZE - tail) - RB_EVNT_HDR_SIZE; 2282 event->type_len = RINGBUF_TYPE_PADDING; 2283 /* time delta must be non zero */ 2284 event->time_delta = 1; 2285 2286 /* Set write to end of buffer */ 2287 length = (tail + length) - BUF_PAGE_SIZE; 2288 local_sub(length, &tail_page->write); 2289 } 2290 2291 /* 2292 * This is the slow path, force gcc not to inline it. 2293 */ 2294 static noinline struct ring_buffer_event * 2295 rb_move_tail(struct ring_buffer_per_cpu *cpu_buffer, 2296 unsigned long length, unsigned long tail, 2297 struct buffer_page *tail_page, u64 ts) 2298 { 2299 struct buffer_page *commit_page = cpu_buffer->commit_page; 2300 struct ring_buffer *buffer = cpu_buffer->buffer; 2301 struct buffer_page *next_page; 2302 int ret; 2303 2304 next_page = tail_page; 2305 2306 rb_inc_page(cpu_buffer, &next_page); 2307 2308 /* 2309 * If for some reason, we had an interrupt storm that made 2310 * it all the way around the buffer, bail, and warn 2311 * about it. 2312 */ 2313 if (unlikely(next_page == commit_page)) { 2314 local_inc(&cpu_buffer->commit_overrun); 2315 goto out_reset; 2316 } 2317 2318 /* 2319 * This is where the fun begins! 2320 * 2321 * We are fighting against races between a reader that 2322 * could be on another CPU trying to swap its reader 2323 * page with the buffer head. 2324 * 2325 * We are also fighting against interrupts coming in and 2326 * moving the head or tail on us as well. 2327 * 2328 * If the next page is the head page then we have filled 2329 * the buffer, unless the commit page is still on the 2330 * reader page. 2331 */ 2332 if (rb_is_head_page(cpu_buffer, next_page, &tail_page->list)) { 2333 2334 /* 2335 * If the commit is not on the reader page, then 2336 * move the header page. 2337 */ 2338 if (!rb_is_reader_page(cpu_buffer->commit_page)) { 2339 /* 2340 * If we are not in overwrite mode, 2341 * this is easy, just stop here. 2342 */ 2343 if (!(buffer->flags & RB_FL_OVERWRITE)) { 2344 local_inc(&cpu_buffer->dropped_events); 2345 goto out_reset; 2346 } 2347 2348 ret = rb_handle_head_page(cpu_buffer, 2349 tail_page, 2350 next_page); 2351 if (ret < 0) 2352 goto out_reset; 2353 if (ret) 2354 goto out_again; 2355 } else { 2356 /* 2357 * We need to be careful here too. The 2358 * commit page could still be on the reader 2359 * page. We could have a small buffer, and 2360 * have filled up the buffer with events 2361 * from interrupts and such, and wrapped. 2362 * 2363 * Note, if the tail page is also the on the 2364 * reader_page, we let it move out. 2365 */ 2366 if (unlikely((cpu_buffer->commit_page != 2367 cpu_buffer->tail_page) && 2368 (cpu_buffer->commit_page == 2369 cpu_buffer->reader_page))) { 2370 local_inc(&cpu_buffer->commit_overrun); 2371 goto out_reset; 2372 } 2373 } 2374 } 2375 2376 ret = rb_tail_page_update(cpu_buffer, tail_page, next_page); 2377 if (ret) { 2378 /* 2379 * Nested commits always have zero deltas, so 2380 * just reread the time stamp 2381 */ 2382 ts = rb_time_stamp(buffer); 2383 next_page->page->time_stamp = ts; 2384 } 2385 2386 out_again: 2387 2388 rb_reset_tail(cpu_buffer, tail_page, tail, length); 2389 2390 /* fail and let the caller try again */ 2391 return ERR_PTR(-EAGAIN); 2392 2393 out_reset: 2394 /* reset write */ 2395 rb_reset_tail(cpu_buffer, tail_page, tail, length); 2396 2397 return NULL; 2398 } 2399 2400 static struct ring_buffer_event * 2401 __rb_reserve_next(struct ring_buffer_per_cpu *cpu_buffer, 2402 unsigned long length, u64 ts, 2403 u64 delta, int add_timestamp) 2404 { 2405 struct buffer_page *tail_page; 2406 struct ring_buffer_event *event; 2407 unsigned long tail, write; 2408 2409 /* 2410 * If the time delta since the last event is too big to 2411 * hold in the time field of the event, then we append a 2412 * TIME EXTEND event ahead of the data event. 2413 */ 2414 if (unlikely(add_timestamp)) 2415 length += RB_LEN_TIME_EXTEND; 2416 2417 tail_page = cpu_buffer->tail_page; 2418 write = local_add_return(length, &tail_page->write); 2419 2420 /* set write to only the index of the write */ 2421 write &= RB_WRITE_MASK; 2422 tail = write - length; 2423 2424 /* 2425 * If this is the first commit on the page, then it has the same 2426 * timestamp as the page itself. 2427 */ 2428 if (!tail) 2429 delta = 0; 2430 2431 /* See if we shot pass the end of this buffer page */ 2432 if (unlikely(write > BUF_PAGE_SIZE)) 2433 return rb_move_tail(cpu_buffer, length, tail, 2434 tail_page, ts); 2435 2436 /* We reserved something on the buffer */ 2437 2438 event = __rb_page_index(tail_page, tail); 2439 kmemcheck_annotate_bitfield(event, bitfield); 2440 rb_update_event(cpu_buffer, event, length, add_timestamp, delta); 2441 2442 local_inc(&tail_page->entries); 2443 2444 /* 2445 * If this is the first commit on the page, then update 2446 * its timestamp. 2447 */ 2448 if (!tail) 2449 tail_page->page->time_stamp = ts; 2450 2451 /* account for these added bytes */ 2452 local_add(length, &cpu_buffer->entries_bytes); 2453 2454 return event; 2455 } 2456 2457 static inline int 2458 rb_try_to_discard(struct ring_buffer_per_cpu *cpu_buffer, 2459 struct ring_buffer_event *event) 2460 { 2461 unsigned long new_index, old_index; 2462 struct buffer_page *bpage; 2463 unsigned long index; 2464 unsigned long addr; 2465 2466 new_index = rb_event_index(event); 2467 old_index = new_index + rb_event_ts_length(event); 2468 addr = (unsigned long)event; 2469 addr &= PAGE_MASK; 2470 2471 bpage = cpu_buffer->tail_page; 2472 2473 if (bpage->page == (void *)addr && rb_page_write(bpage) == old_index) { 2474 unsigned long write_mask = 2475 local_read(&bpage->write) & ~RB_WRITE_MASK; 2476 unsigned long event_length = rb_event_length(event); 2477 /* 2478 * This is on the tail page. It is possible that 2479 * a write could come in and move the tail page 2480 * and write to the next page. That is fine 2481 * because we just shorten what is on this page. 2482 */ 2483 old_index += write_mask; 2484 new_index += write_mask; 2485 index = local_cmpxchg(&bpage->write, old_index, new_index); 2486 if (index == old_index) { 2487 /* update counters */ 2488 local_sub(event_length, &cpu_buffer->entries_bytes); 2489 return 1; 2490 } 2491 } 2492 2493 /* could not discard */ 2494 return 0; 2495 } 2496 2497 static void rb_start_commit(struct ring_buffer_per_cpu *cpu_buffer) 2498 { 2499 local_inc(&cpu_buffer->committing); 2500 local_inc(&cpu_buffer->commits); 2501 } 2502 2503 static inline void rb_end_commit(struct ring_buffer_per_cpu *cpu_buffer) 2504 { 2505 unsigned long commits; 2506 2507 if (RB_WARN_ON(cpu_buffer, 2508 !local_read(&cpu_buffer->committing))) 2509 return; 2510 2511 again: 2512 commits = local_read(&cpu_buffer->commits); 2513 /* synchronize with interrupts */ 2514 barrier(); 2515 if (local_read(&cpu_buffer->committing) == 1) 2516 rb_set_commit_to_write(cpu_buffer); 2517 2518 local_dec(&cpu_buffer->committing); 2519 2520 /* synchronize with interrupts */ 2521 barrier(); 2522 2523 /* 2524 * Need to account for interrupts coming in between the 2525 * updating of the commit page and the clearing of the 2526 * committing counter. 2527 */ 2528 if (unlikely(local_read(&cpu_buffer->commits) != commits) && 2529 !local_read(&cpu_buffer->committing)) { 2530 local_inc(&cpu_buffer->committing); 2531 goto again; 2532 } 2533 } 2534 2535 static struct ring_buffer_event * 2536 rb_reserve_next_event(struct ring_buffer *buffer, 2537 struct ring_buffer_per_cpu *cpu_buffer, 2538 unsigned long length) 2539 { 2540 struct ring_buffer_event *event; 2541 u64 ts, delta; 2542 int nr_loops = 0; 2543 int add_timestamp; 2544 u64 diff; 2545 2546 rb_start_commit(cpu_buffer); 2547 2548 #ifdef CONFIG_RING_BUFFER_ALLOW_SWAP 2549 /* 2550 * Due to the ability to swap a cpu buffer from a buffer 2551 * it is possible it was swapped before we committed. 2552 * (committing stops a swap). We check for it here and 2553 * if it happened, we have to fail the write. 2554 */ 2555 barrier(); 2556 if (unlikely(ACCESS_ONCE(cpu_buffer->buffer) != buffer)) { 2557 local_dec(&cpu_buffer->committing); 2558 local_dec(&cpu_buffer->commits); 2559 return NULL; 2560 } 2561 #endif 2562 2563 length = rb_calculate_event_length(length); 2564 again: 2565 add_timestamp = 0; 2566 delta = 0; 2567 2568 /* 2569 * We allow for interrupts to reenter here and do a trace. 2570 * If one does, it will cause this original code to loop 2571 * back here. Even with heavy interrupts happening, this 2572 * should only happen a few times in a row. If this happens 2573 * 1000 times in a row, there must be either an interrupt 2574 * storm or we have something buggy. 2575 * Bail! 2576 */ 2577 if (RB_WARN_ON(cpu_buffer, ++nr_loops > 1000)) 2578 goto out_fail; 2579 2580 ts = rb_time_stamp(cpu_buffer->buffer); 2581 diff = ts - cpu_buffer->write_stamp; 2582 2583 /* make sure this diff is calculated here */ 2584 barrier(); 2585 2586 /* Did the write stamp get updated already? */ 2587 if (likely(ts >= cpu_buffer->write_stamp)) { 2588 delta = diff; 2589 if (unlikely(test_time_stamp(delta))) { 2590 int local_clock_stable = 1; 2591 #ifdef CONFIG_HAVE_UNSTABLE_SCHED_CLOCK 2592 local_clock_stable = sched_clock_stable(); 2593 #endif 2594 WARN_ONCE(delta > (1ULL << 59), 2595 KERN_WARNING "Delta way too big! %llu ts=%llu write stamp = %llu\n%s", 2596 (unsigned long long)delta, 2597 (unsigned long long)ts, 2598 (unsigned long long)cpu_buffer->write_stamp, 2599 local_clock_stable ? "" : 2600 "If you just came from a suspend/resume,\n" 2601 "please switch to the trace global clock:\n" 2602 " echo global > /sys/kernel/debug/tracing/trace_clock\n"); 2603 add_timestamp = 1; 2604 } 2605 } 2606 2607 event = __rb_reserve_next(cpu_buffer, length, ts, 2608 delta, add_timestamp); 2609 if (unlikely(PTR_ERR(event) == -EAGAIN)) 2610 goto again; 2611 2612 if (!event) 2613 goto out_fail; 2614 2615 return event; 2616 2617 out_fail: 2618 rb_end_commit(cpu_buffer); 2619 return NULL; 2620 } 2621 2622 #ifdef CONFIG_TRACING 2623 2624 /* 2625 * The lock and unlock are done within a preempt disable section. 2626 * The current_context per_cpu variable can only be modified 2627 * by the current task between lock and unlock. But it can 2628 * be modified more than once via an interrupt. To pass this 2629 * information from the lock to the unlock without having to 2630 * access the 'in_interrupt()' functions again (which do show 2631 * a bit of overhead in something as critical as function tracing, 2632 * we use a bitmask trick. 2633 * 2634 * bit 0 = NMI context 2635 * bit 1 = IRQ context 2636 * bit 2 = SoftIRQ context 2637 * bit 3 = normal context. 2638 * 2639 * This works because this is the order of contexts that can 2640 * preempt other contexts. A SoftIRQ never preempts an IRQ 2641 * context. 2642 * 2643 * When the context is determined, the corresponding bit is 2644 * checked and set (if it was set, then a recursion of that context 2645 * happened). 2646 * 2647 * On unlock, we need to clear this bit. To do so, just subtract 2648 * 1 from the current_context and AND it to itself. 2649 * 2650 * (binary) 2651 * 101 - 1 = 100 2652 * 101 & 100 = 100 (clearing bit zero) 2653 * 2654 * 1010 - 1 = 1001 2655 * 1010 & 1001 = 1000 (clearing bit 1) 2656 * 2657 * The least significant bit can be cleared this way, and it 2658 * just so happens that it is the same bit corresponding to 2659 * the current context. 2660 */ 2661 static DEFINE_PER_CPU(unsigned int, current_context); 2662 2663 static __always_inline int trace_recursive_lock(void) 2664 { 2665 unsigned int val = this_cpu_read(current_context); 2666 int bit; 2667 2668 if (in_interrupt()) { 2669 if (in_nmi()) 2670 bit = 0; 2671 else if (in_irq()) 2672 bit = 1; 2673 else 2674 bit = 2; 2675 } else 2676 bit = 3; 2677 2678 if (unlikely(val & (1 << bit))) 2679 return 1; 2680 2681 val |= (1 << bit); 2682 this_cpu_write(current_context, val); 2683 2684 return 0; 2685 } 2686 2687 static __always_inline void trace_recursive_unlock(void) 2688 { 2689 unsigned int val = this_cpu_read(current_context); 2690 2691 val--; 2692 val &= this_cpu_read(current_context); 2693 this_cpu_write(current_context, val); 2694 } 2695 2696 #else 2697 2698 #define trace_recursive_lock() (0) 2699 #define trace_recursive_unlock() do { } while (0) 2700 2701 #endif 2702 2703 /** 2704 * ring_buffer_lock_reserve - reserve a part of the buffer 2705 * @buffer: the ring buffer to reserve from 2706 * @length: the length of the data to reserve (excluding event header) 2707 * 2708 * Returns a reseverd event on the ring buffer to copy directly to. 2709 * The user of this interface will need to get the body to write into 2710 * and can use the ring_buffer_event_data() interface. 2711 * 2712 * The length is the length of the data needed, not the event length 2713 * which also includes the event header. 2714 * 2715 * Must be paired with ring_buffer_unlock_commit, unless NULL is returned. 2716 * If NULL is returned, then nothing has been allocated or locked. 2717 */ 2718 struct ring_buffer_event * 2719 ring_buffer_lock_reserve(struct ring_buffer *buffer, unsigned long length) 2720 { 2721 struct ring_buffer_per_cpu *cpu_buffer; 2722 struct ring_buffer_event *event; 2723 int cpu; 2724 2725 if (ring_buffer_flags != RB_BUFFERS_ON) 2726 return NULL; 2727 2728 /* If we are tracing schedule, we don't want to recurse */ 2729 preempt_disable_notrace(); 2730 2731 if (atomic_read(&buffer->record_disabled)) 2732 goto out_nocheck; 2733 2734 if (trace_recursive_lock()) 2735 goto out_nocheck; 2736 2737 cpu = raw_smp_processor_id(); 2738 2739 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 2740 goto out; 2741 2742 cpu_buffer = buffer->buffers[cpu]; 2743 2744 if (atomic_read(&cpu_buffer->record_disabled)) 2745 goto out; 2746 2747 if (length > BUF_MAX_DATA_SIZE) 2748 goto out; 2749 2750 event = rb_reserve_next_event(buffer, cpu_buffer, length); 2751 if (!event) 2752 goto out; 2753 2754 return event; 2755 2756 out: 2757 trace_recursive_unlock(); 2758 2759 out_nocheck: 2760 preempt_enable_notrace(); 2761 return NULL; 2762 } 2763 EXPORT_SYMBOL_GPL(ring_buffer_lock_reserve); 2764 2765 static void 2766 rb_update_write_stamp(struct ring_buffer_per_cpu *cpu_buffer, 2767 struct ring_buffer_event *event) 2768 { 2769 u64 delta; 2770 2771 /* 2772 * The event first in the commit queue updates the 2773 * time stamp. 2774 */ 2775 if (rb_event_is_commit(cpu_buffer, event)) { 2776 /* 2777 * A commit event that is first on a page 2778 * updates the write timestamp with the page stamp 2779 */ 2780 if (!rb_event_index(event)) 2781 cpu_buffer->write_stamp = 2782 cpu_buffer->commit_page->page->time_stamp; 2783 else if (event->type_len == RINGBUF_TYPE_TIME_EXTEND) { 2784 delta = event->array[0]; 2785 delta <<= TS_SHIFT; 2786 delta += event->time_delta; 2787 cpu_buffer->write_stamp += delta; 2788 } else 2789 cpu_buffer->write_stamp += event->time_delta; 2790 } 2791 } 2792 2793 static void rb_commit(struct ring_buffer_per_cpu *cpu_buffer, 2794 struct ring_buffer_event *event) 2795 { 2796 local_inc(&cpu_buffer->entries); 2797 rb_update_write_stamp(cpu_buffer, event); 2798 rb_end_commit(cpu_buffer); 2799 } 2800 2801 static __always_inline void 2802 rb_wakeups(struct ring_buffer *buffer, struct ring_buffer_per_cpu *cpu_buffer) 2803 { 2804 if (buffer->irq_work.waiters_pending) { 2805 buffer->irq_work.waiters_pending = false; 2806 /* irq_work_queue() supplies it's own memory barriers */ 2807 irq_work_queue(&buffer->irq_work.work); 2808 } 2809 2810 if (cpu_buffer->irq_work.waiters_pending) { 2811 cpu_buffer->irq_work.waiters_pending = false; 2812 /* irq_work_queue() supplies it's own memory barriers */ 2813 irq_work_queue(&cpu_buffer->irq_work.work); 2814 } 2815 } 2816 2817 /** 2818 * ring_buffer_unlock_commit - commit a reserved 2819 * @buffer: The buffer to commit to 2820 * @event: The event pointer to commit. 2821 * 2822 * This commits the data to the ring buffer, and releases any locks held. 2823 * 2824 * Must be paired with ring_buffer_lock_reserve. 2825 */ 2826 int ring_buffer_unlock_commit(struct ring_buffer *buffer, 2827 struct ring_buffer_event *event) 2828 { 2829 struct ring_buffer_per_cpu *cpu_buffer; 2830 int cpu = raw_smp_processor_id(); 2831 2832 cpu_buffer = buffer->buffers[cpu]; 2833 2834 rb_commit(cpu_buffer, event); 2835 2836 rb_wakeups(buffer, cpu_buffer); 2837 2838 trace_recursive_unlock(); 2839 2840 preempt_enable_notrace(); 2841 2842 return 0; 2843 } 2844 EXPORT_SYMBOL_GPL(ring_buffer_unlock_commit); 2845 2846 static inline void rb_event_discard(struct ring_buffer_event *event) 2847 { 2848 if (event->type_len == RINGBUF_TYPE_TIME_EXTEND) 2849 event = skip_time_extend(event); 2850 2851 /* array[0] holds the actual length for the discarded event */ 2852 event->array[0] = rb_event_data_length(event) - RB_EVNT_HDR_SIZE; 2853 event->type_len = RINGBUF_TYPE_PADDING; 2854 /* time delta must be non zero */ 2855 if (!event->time_delta) 2856 event->time_delta = 1; 2857 } 2858 2859 /* 2860 * Decrement the entries to the page that an event is on. 2861 * The event does not even need to exist, only the pointer 2862 * to the page it is on. This may only be called before the commit 2863 * takes place. 2864 */ 2865 static inline void 2866 rb_decrement_entry(struct ring_buffer_per_cpu *cpu_buffer, 2867 struct ring_buffer_event *event) 2868 { 2869 unsigned long addr = (unsigned long)event; 2870 struct buffer_page *bpage = cpu_buffer->commit_page; 2871 struct buffer_page *start; 2872 2873 addr &= PAGE_MASK; 2874 2875 /* Do the likely case first */ 2876 if (likely(bpage->page == (void *)addr)) { 2877 local_dec(&bpage->entries); 2878 return; 2879 } 2880 2881 /* 2882 * Because the commit page may be on the reader page we 2883 * start with the next page and check the end loop there. 2884 */ 2885 rb_inc_page(cpu_buffer, &bpage); 2886 start = bpage; 2887 do { 2888 if (bpage->page == (void *)addr) { 2889 local_dec(&bpage->entries); 2890 return; 2891 } 2892 rb_inc_page(cpu_buffer, &bpage); 2893 } while (bpage != start); 2894 2895 /* commit not part of this buffer?? */ 2896 RB_WARN_ON(cpu_buffer, 1); 2897 } 2898 2899 /** 2900 * ring_buffer_commit_discard - discard an event that has not been committed 2901 * @buffer: the ring buffer 2902 * @event: non committed event to discard 2903 * 2904 * Sometimes an event that is in the ring buffer needs to be ignored. 2905 * This function lets the user discard an event in the ring buffer 2906 * and then that event will not be read later. 2907 * 2908 * This function only works if it is called before the the item has been 2909 * committed. It will try to free the event from the ring buffer 2910 * if another event has not been added behind it. 2911 * 2912 * If another event has been added behind it, it will set the event 2913 * up as discarded, and perform the commit. 2914 * 2915 * If this function is called, do not call ring_buffer_unlock_commit on 2916 * the event. 2917 */ 2918 void ring_buffer_discard_commit(struct ring_buffer *buffer, 2919 struct ring_buffer_event *event) 2920 { 2921 struct ring_buffer_per_cpu *cpu_buffer; 2922 int cpu; 2923 2924 /* The event is discarded regardless */ 2925 rb_event_discard(event); 2926 2927 cpu = smp_processor_id(); 2928 cpu_buffer = buffer->buffers[cpu]; 2929 2930 /* 2931 * This must only be called if the event has not been 2932 * committed yet. Thus we can assume that preemption 2933 * is still disabled. 2934 */ 2935 RB_WARN_ON(buffer, !local_read(&cpu_buffer->committing)); 2936 2937 rb_decrement_entry(cpu_buffer, event); 2938 if (rb_try_to_discard(cpu_buffer, event)) 2939 goto out; 2940 2941 /* 2942 * The commit is still visible by the reader, so we 2943 * must still update the timestamp. 2944 */ 2945 rb_update_write_stamp(cpu_buffer, event); 2946 out: 2947 rb_end_commit(cpu_buffer); 2948 2949 trace_recursive_unlock(); 2950 2951 preempt_enable_notrace(); 2952 2953 } 2954 EXPORT_SYMBOL_GPL(ring_buffer_discard_commit); 2955 2956 /** 2957 * ring_buffer_write - write data to the buffer without reserving 2958 * @buffer: The ring buffer to write to. 2959 * @length: The length of the data being written (excluding the event header) 2960 * @data: The data to write to the buffer. 2961 * 2962 * This is like ring_buffer_lock_reserve and ring_buffer_unlock_commit as 2963 * one function. If you already have the data to write to the buffer, it 2964 * may be easier to simply call this function. 2965 * 2966 * Note, like ring_buffer_lock_reserve, the length is the length of the data 2967 * and not the length of the event which would hold the header. 2968 */ 2969 int ring_buffer_write(struct ring_buffer *buffer, 2970 unsigned long length, 2971 void *data) 2972 { 2973 struct ring_buffer_per_cpu *cpu_buffer; 2974 struct ring_buffer_event *event; 2975 void *body; 2976 int ret = -EBUSY; 2977 int cpu; 2978 2979 if (ring_buffer_flags != RB_BUFFERS_ON) 2980 return -EBUSY; 2981 2982 preempt_disable_notrace(); 2983 2984 if (atomic_read(&buffer->record_disabled)) 2985 goto out; 2986 2987 cpu = raw_smp_processor_id(); 2988 2989 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 2990 goto out; 2991 2992 cpu_buffer = buffer->buffers[cpu]; 2993 2994 if (atomic_read(&cpu_buffer->record_disabled)) 2995 goto out; 2996 2997 if (length > BUF_MAX_DATA_SIZE) 2998 goto out; 2999 3000 event = rb_reserve_next_event(buffer, cpu_buffer, length); 3001 if (!event) 3002 goto out; 3003 3004 body = rb_event_data(event); 3005 3006 memcpy(body, data, length); 3007 3008 rb_commit(cpu_buffer, event); 3009 3010 rb_wakeups(buffer, cpu_buffer); 3011 3012 ret = 0; 3013 out: 3014 preempt_enable_notrace(); 3015 3016 return ret; 3017 } 3018 EXPORT_SYMBOL_GPL(ring_buffer_write); 3019 3020 static int rb_per_cpu_empty(struct ring_buffer_per_cpu *cpu_buffer) 3021 { 3022 struct buffer_page *reader = cpu_buffer->reader_page; 3023 struct buffer_page *head = rb_set_head_page(cpu_buffer); 3024 struct buffer_page *commit = cpu_buffer->commit_page; 3025 3026 /* In case of error, head will be NULL */ 3027 if (unlikely(!head)) 3028 return 1; 3029 3030 return reader->read == rb_page_commit(reader) && 3031 (commit == reader || 3032 (commit == head && 3033 head->read == rb_page_commit(commit))); 3034 } 3035 3036 /** 3037 * ring_buffer_record_disable - stop all writes into the buffer 3038 * @buffer: The ring buffer to stop writes to. 3039 * 3040 * This prevents all writes to the buffer. Any attempt to write 3041 * to the buffer after this will fail and return NULL. 3042 * 3043 * The caller should call synchronize_sched() after this. 3044 */ 3045 void ring_buffer_record_disable(struct ring_buffer *buffer) 3046 { 3047 atomic_inc(&buffer->record_disabled); 3048 } 3049 EXPORT_SYMBOL_GPL(ring_buffer_record_disable); 3050 3051 /** 3052 * ring_buffer_record_enable - enable writes to the buffer 3053 * @buffer: The ring buffer to enable writes 3054 * 3055 * Note, multiple disables will need the same number of enables 3056 * to truly enable the writing (much like preempt_disable). 3057 */ 3058 void ring_buffer_record_enable(struct ring_buffer *buffer) 3059 { 3060 atomic_dec(&buffer->record_disabled); 3061 } 3062 EXPORT_SYMBOL_GPL(ring_buffer_record_enable); 3063 3064 /** 3065 * ring_buffer_record_off - stop all writes into the buffer 3066 * @buffer: The ring buffer to stop writes to. 3067 * 3068 * This prevents all writes to the buffer. Any attempt to write 3069 * to the buffer after this will fail and return NULL. 3070 * 3071 * This is different than ring_buffer_record_disable() as 3072 * it works like an on/off switch, where as the disable() version 3073 * must be paired with a enable(). 3074 */ 3075 void ring_buffer_record_off(struct ring_buffer *buffer) 3076 { 3077 unsigned int rd; 3078 unsigned int new_rd; 3079 3080 do { 3081 rd = atomic_read(&buffer->record_disabled); 3082 new_rd = rd | RB_BUFFER_OFF; 3083 } while (atomic_cmpxchg(&buffer->record_disabled, rd, new_rd) != rd); 3084 } 3085 EXPORT_SYMBOL_GPL(ring_buffer_record_off); 3086 3087 /** 3088 * ring_buffer_record_on - restart writes into the buffer 3089 * @buffer: The ring buffer to start writes to. 3090 * 3091 * This enables all writes to the buffer that was disabled by 3092 * ring_buffer_record_off(). 3093 * 3094 * This is different than ring_buffer_record_enable() as 3095 * it works like an on/off switch, where as the enable() version 3096 * must be paired with a disable(). 3097 */ 3098 void ring_buffer_record_on(struct ring_buffer *buffer) 3099 { 3100 unsigned int rd; 3101 unsigned int new_rd; 3102 3103 do { 3104 rd = atomic_read(&buffer->record_disabled); 3105 new_rd = rd & ~RB_BUFFER_OFF; 3106 } while (atomic_cmpxchg(&buffer->record_disabled, rd, new_rd) != rd); 3107 } 3108 EXPORT_SYMBOL_GPL(ring_buffer_record_on); 3109 3110 /** 3111 * ring_buffer_record_is_on - return true if the ring buffer can write 3112 * @buffer: The ring buffer to see if write is enabled 3113 * 3114 * Returns true if the ring buffer is in a state that it accepts writes. 3115 */ 3116 int ring_buffer_record_is_on(struct ring_buffer *buffer) 3117 { 3118 return !atomic_read(&buffer->record_disabled); 3119 } 3120 3121 /** 3122 * ring_buffer_record_disable_cpu - stop all writes into the cpu_buffer 3123 * @buffer: The ring buffer to stop writes to. 3124 * @cpu: The CPU buffer to stop 3125 * 3126 * This prevents all writes to the buffer. Any attempt to write 3127 * to the buffer after this will fail and return NULL. 3128 * 3129 * The caller should call synchronize_sched() after this. 3130 */ 3131 void ring_buffer_record_disable_cpu(struct ring_buffer *buffer, int cpu) 3132 { 3133 struct ring_buffer_per_cpu *cpu_buffer; 3134 3135 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 3136 return; 3137 3138 cpu_buffer = buffer->buffers[cpu]; 3139 atomic_inc(&cpu_buffer->record_disabled); 3140 } 3141 EXPORT_SYMBOL_GPL(ring_buffer_record_disable_cpu); 3142 3143 /** 3144 * ring_buffer_record_enable_cpu - enable writes to the buffer 3145 * @buffer: The ring buffer to enable writes 3146 * @cpu: The CPU to enable. 3147 * 3148 * Note, multiple disables will need the same number of enables 3149 * to truly enable the writing (much like preempt_disable). 3150 */ 3151 void ring_buffer_record_enable_cpu(struct ring_buffer *buffer, int cpu) 3152 { 3153 struct ring_buffer_per_cpu *cpu_buffer; 3154 3155 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 3156 return; 3157 3158 cpu_buffer = buffer->buffers[cpu]; 3159 atomic_dec(&cpu_buffer->record_disabled); 3160 } 3161 EXPORT_SYMBOL_GPL(ring_buffer_record_enable_cpu); 3162 3163 /* 3164 * The total entries in the ring buffer is the running counter 3165 * of entries entered into the ring buffer, minus the sum of 3166 * the entries read from the ring buffer and the number of 3167 * entries that were overwritten. 3168 */ 3169 static inline unsigned long 3170 rb_num_of_entries(struct ring_buffer_per_cpu *cpu_buffer) 3171 { 3172 return local_read(&cpu_buffer->entries) - 3173 (local_read(&cpu_buffer->overrun) + cpu_buffer->read); 3174 } 3175 3176 /** 3177 * ring_buffer_oldest_event_ts - get the oldest event timestamp from the buffer 3178 * @buffer: The ring buffer 3179 * @cpu: The per CPU buffer to read from. 3180 */ 3181 u64 ring_buffer_oldest_event_ts(struct ring_buffer *buffer, int cpu) 3182 { 3183 unsigned long flags; 3184 struct ring_buffer_per_cpu *cpu_buffer; 3185 struct buffer_page *bpage; 3186 u64 ret = 0; 3187 3188 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 3189 return 0; 3190 3191 cpu_buffer = buffer->buffers[cpu]; 3192 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 3193 /* 3194 * if the tail is on reader_page, oldest time stamp is on the reader 3195 * page 3196 */ 3197 if (cpu_buffer->tail_page == cpu_buffer->reader_page) 3198 bpage = cpu_buffer->reader_page; 3199 else 3200 bpage = rb_set_head_page(cpu_buffer); 3201 if (bpage) 3202 ret = bpage->page->time_stamp; 3203 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 3204 3205 return ret; 3206 } 3207 EXPORT_SYMBOL_GPL(ring_buffer_oldest_event_ts); 3208 3209 /** 3210 * ring_buffer_bytes_cpu - get the number of bytes consumed in a cpu buffer 3211 * @buffer: The ring buffer 3212 * @cpu: The per CPU buffer to read from. 3213 */ 3214 unsigned long ring_buffer_bytes_cpu(struct ring_buffer *buffer, int cpu) 3215 { 3216 struct ring_buffer_per_cpu *cpu_buffer; 3217 unsigned long ret; 3218 3219 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 3220 return 0; 3221 3222 cpu_buffer = buffer->buffers[cpu]; 3223 ret = local_read(&cpu_buffer->entries_bytes) - cpu_buffer->read_bytes; 3224 3225 return ret; 3226 } 3227 EXPORT_SYMBOL_GPL(ring_buffer_bytes_cpu); 3228 3229 /** 3230 * ring_buffer_entries_cpu - get the number of entries in a cpu buffer 3231 * @buffer: The ring buffer 3232 * @cpu: The per CPU buffer to get the entries from. 3233 */ 3234 unsigned long ring_buffer_entries_cpu(struct ring_buffer *buffer, int cpu) 3235 { 3236 struct ring_buffer_per_cpu *cpu_buffer; 3237 3238 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 3239 return 0; 3240 3241 cpu_buffer = buffer->buffers[cpu]; 3242 3243 return rb_num_of_entries(cpu_buffer); 3244 } 3245 EXPORT_SYMBOL_GPL(ring_buffer_entries_cpu); 3246 3247 /** 3248 * ring_buffer_overrun_cpu - get the number of overruns caused by the ring 3249 * buffer wrapping around (only if RB_FL_OVERWRITE is on). 3250 * @buffer: The ring buffer 3251 * @cpu: The per CPU buffer to get the number of overruns from 3252 */ 3253 unsigned long ring_buffer_overrun_cpu(struct ring_buffer *buffer, int cpu) 3254 { 3255 struct ring_buffer_per_cpu *cpu_buffer; 3256 unsigned long ret; 3257 3258 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 3259 return 0; 3260 3261 cpu_buffer = buffer->buffers[cpu]; 3262 ret = local_read(&cpu_buffer->overrun); 3263 3264 return ret; 3265 } 3266 EXPORT_SYMBOL_GPL(ring_buffer_overrun_cpu); 3267 3268 /** 3269 * ring_buffer_commit_overrun_cpu - get the number of overruns caused by 3270 * commits failing due to the buffer wrapping around while there are uncommitted 3271 * events, such as during an interrupt storm. 3272 * @buffer: The ring buffer 3273 * @cpu: The per CPU buffer to get the number of overruns from 3274 */ 3275 unsigned long 3276 ring_buffer_commit_overrun_cpu(struct ring_buffer *buffer, int cpu) 3277 { 3278 struct ring_buffer_per_cpu *cpu_buffer; 3279 unsigned long ret; 3280 3281 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 3282 return 0; 3283 3284 cpu_buffer = buffer->buffers[cpu]; 3285 ret = local_read(&cpu_buffer->commit_overrun); 3286 3287 return ret; 3288 } 3289 EXPORT_SYMBOL_GPL(ring_buffer_commit_overrun_cpu); 3290 3291 /** 3292 * ring_buffer_dropped_events_cpu - get the number of dropped events caused by 3293 * the ring buffer filling up (only if RB_FL_OVERWRITE is off). 3294 * @buffer: The ring buffer 3295 * @cpu: The per CPU buffer to get the number of overruns from 3296 */ 3297 unsigned long 3298 ring_buffer_dropped_events_cpu(struct ring_buffer *buffer, int cpu) 3299 { 3300 struct ring_buffer_per_cpu *cpu_buffer; 3301 unsigned long ret; 3302 3303 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 3304 return 0; 3305 3306 cpu_buffer = buffer->buffers[cpu]; 3307 ret = local_read(&cpu_buffer->dropped_events); 3308 3309 return ret; 3310 } 3311 EXPORT_SYMBOL_GPL(ring_buffer_dropped_events_cpu); 3312 3313 /** 3314 * ring_buffer_read_events_cpu - get the number of events successfully read 3315 * @buffer: The ring buffer 3316 * @cpu: The per CPU buffer to get the number of events read 3317 */ 3318 unsigned long 3319 ring_buffer_read_events_cpu(struct ring_buffer *buffer, int cpu) 3320 { 3321 struct ring_buffer_per_cpu *cpu_buffer; 3322 3323 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 3324 return 0; 3325 3326 cpu_buffer = buffer->buffers[cpu]; 3327 return cpu_buffer->read; 3328 } 3329 EXPORT_SYMBOL_GPL(ring_buffer_read_events_cpu); 3330 3331 /** 3332 * ring_buffer_entries - get the number of entries in a buffer 3333 * @buffer: The ring buffer 3334 * 3335 * Returns the total number of entries in the ring buffer 3336 * (all CPU entries) 3337 */ 3338 unsigned long ring_buffer_entries(struct ring_buffer *buffer) 3339 { 3340 struct ring_buffer_per_cpu *cpu_buffer; 3341 unsigned long entries = 0; 3342 int cpu; 3343 3344 /* if you care about this being correct, lock the buffer */ 3345 for_each_buffer_cpu(buffer, cpu) { 3346 cpu_buffer = buffer->buffers[cpu]; 3347 entries += rb_num_of_entries(cpu_buffer); 3348 } 3349 3350 return entries; 3351 } 3352 EXPORT_SYMBOL_GPL(ring_buffer_entries); 3353 3354 /** 3355 * ring_buffer_overruns - get the number of overruns in buffer 3356 * @buffer: The ring buffer 3357 * 3358 * Returns the total number of overruns in the ring buffer 3359 * (all CPU entries) 3360 */ 3361 unsigned long ring_buffer_overruns(struct ring_buffer *buffer) 3362 { 3363 struct ring_buffer_per_cpu *cpu_buffer; 3364 unsigned long overruns = 0; 3365 int cpu; 3366 3367 /* if you care about this being correct, lock the buffer */ 3368 for_each_buffer_cpu(buffer, cpu) { 3369 cpu_buffer = buffer->buffers[cpu]; 3370 overruns += local_read(&cpu_buffer->overrun); 3371 } 3372 3373 return overruns; 3374 } 3375 EXPORT_SYMBOL_GPL(ring_buffer_overruns); 3376 3377 static void rb_iter_reset(struct ring_buffer_iter *iter) 3378 { 3379 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer; 3380 3381 /* Iterator usage is expected to have record disabled */ 3382 iter->head_page = cpu_buffer->reader_page; 3383 iter->head = cpu_buffer->reader_page->read; 3384 3385 iter->cache_reader_page = iter->head_page; 3386 iter->cache_read = cpu_buffer->read; 3387 3388 if (iter->head) 3389 iter->read_stamp = cpu_buffer->read_stamp; 3390 else 3391 iter->read_stamp = iter->head_page->page->time_stamp; 3392 } 3393 3394 /** 3395 * ring_buffer_iter_reset - reset an iterator 3396 * @iter: The iterator to reset 3397 * 3398 * Resets the iterator, so that it will start from the beginning 3399 * again. 3400 */ 3401 void ring_buffer_iter_reset(struct ring_buffer_iter *iter) 3402 { 3403 struct ring_buffer_per_cpu *cpu_buffer; 3404 unsigned long flags; 3405 3406 if (!iter) 3407 return; 3408 3409 cpu_buffer = iter->cpu_buffer; 3410 3411 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 3412 rb_iter_reset(iter); 3413 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 3414 } 3415 EXPORT_SYMBOL_GPL(ring_buffer_iter_reset); 3416 3417 /** 3418 * ring_buffer_iter_empty - check if an iterator has no more to read 3419 * @iter: The iterator to check 3420 */ 3421 int ring_buffer_iter_empty(struct ring_buffer_iter *iter) 3422 { 3423 struct ring_buffer_per_cpu *cpu_buffer; 3424 3425 cpu_buffer = iter->cpu_buffer; 3426 3427 return iter->head_page == cpu_buffer->commit_page && 3428 iter->head == rb_commit_index(cpu_buffer); 3429 } 3430 EXPORT_SYMBOL_GPL(ring_buffer_iter_empty); 3431 3432 static void 3433 rb_update_read_stamp(struct ring_buffer_per_cpu *cpu_buffer, 3434 struct ring_buffer_event *event) 3435 { 3436 u64 delta; 3437 3438 switch (event->type_len) { 3439 case RINGBUF_TYPE_PADDING: 3440 return; 3441 3442 case RINGBUF_TYPE_TIME_EXTEND: 3443 delta = event->array[0]; 3444 delta <<= TS_SHIFT; 3445 delta += event->time_delta; 3446 cpu_buffer->read_stamp += delta; 3447 return; 3448 3449 case RINGBUF_TYPE_TIME_STAMP: 3450 /* FIXME: not implemented */ 3451 return; 3452 3453 case RINGBUF_TYPE_DATA: 3454 cpu_buffer->read_stamp += event->time_delta; 3455 return; 3456 3457 default: 3458 BUG(); 3459 } 3460 return; 3461 } 3462 3463 static void 3464 rb_update_iter_read_stamp(struct ring_buffer_iter *iter, 3465 struct ring_buffer_event *event) 3466 { 3467 u64 delta; 3468 3469 switch (event->type_len) { 3470 case RINGBUF_TYPE_PADDING: 3471 return; 3472 3473 case RINGBUF_TYPE_TIME_EXTEND: 3474 delta = event->array[0]; 3475 delta <<= TS_SHIFT; 3476 delta += event->time_delta; 3477 iter->read_stamp += delta; 3478 return; 3479 3480 case RINGBUF_TYPE_TIME_STAMP: 3481 /* FIXME: not implemented */ 3482 return; 3483 3484 case RINGBUF_TYPE_DATA: 3485 iter->read_stamp += event->time_delta; 3486 return; 3487 3488 default: 3489 BUG(); 3490 } 3491 return; 3492 } 3493 3494 static struct buffer_page * 3495 rb_get_reader_page(struct ring_buffer_per_cpu *cpu_buffer) 3496 { 3497 struct buffer_page *reader = NULL; 3498 unsigned long overwrite; 3499 unsigned long flags; 3500 int nr_loops = 0; 3501 int ret; 3502 3503 local_irq_save(flags); 3504 arch_spin_lock(&cpu_buffer->lock); 3505 3506 again: 3507 /* 3508 * This should normally only loop twice. But because the 3509 * start of the reader inserts an empty page, it causes 3510 * a case where we will loop three times. There should be no 3511 * reason to loop four times (that I know of). 3512 */ 3513 if (RB_WARN_ON(cpu_buffer, ++nr_loops > 3)) { 3514 reader = NULL; 3515 goto out; 3516 } 3517 3518 reader = cpu_buffer->reader_page; 3519 3520 /* If there's more to read, return this page */ 3521 if (cpu_buffer->reader_page->read < rb_page_size(reader)) 3522 goto out; 3523 3524 /* Never should we have an index greater than the size */ 3525 if (RB_WARN_ON(cpu_buffer, 3526 cpu_buffer->reader_page->read > rb_page_size(reader))) 3527 goto out; 3528 3529 /* check if we caught up to the tail */ 3530 reader = NULL; 3531 if (cpu_buffer->commit_page == cpu_buffer->reader_page) 3532 goto out; 3533 3534 /* Don't bother swapping if the ring buffer is empty */ 3535 if (rb_num_of_entries(cpu_buffer) == 0) 3536 goto out; 3537 3538 /* 3539 * Reset the reader page to size zero. 3540 */ 3541 local_set(&cpu_buffer->reader_page->write, 0); 3542 local_set(&cpu_buffer->reader_page->entries, 0); 3543 local_set(&cpu_buffer->reader_page->page->commit, 0); 3544 cpu_buffer->reader_page->real_end = 0; 3545 3546 spin: 3547 /* 3548 * Splice the empty reader page into the list around the head. 3549 */ 3550 reader = rb_set_head_page(cpu_buffer); 3551 if (!reader) 3552 goto out; 3553 cpu_buffer->reader_page->list.next = rb_list_head(reader->list.next); 3554 cpu_buffer->reader_page->list.prev = reader->list.prev; 3555 3556 /* 3557 * cpu_buffer->pages just needs to point to the buffer, it 3558 * has no specific buffer page to point to. Lets move it out 3559 * of our way so we don't accidentally swap it. 3560 */ 3561 cpu_buffer->pages = reader->list.prev; 3562 3563 /* The reader page will be pointing to the new head */ 3564 rb_set_list_to_head(cpu_buffer, &cpu_buffer->reader_page->list); 3565 3566 /* 3567 * We want to make sure we read the overruns after we set up our 3568 * pointers to the next object. The writer side does a 3569 * cmpxchg to cross pages which acts as the mb on the writer 3570 * side. Note, the reader will constantly fail the swap 3571 * while the writer is updating the pointers, so this 3572 * guarantees that the overwrite recorded here is the one we 3573 * want to compare with the last_overrun. 3574 */ 3575 smp_mb(); 3576 overwrite = local_read(&(cpu_buffer->overrun)); 3577 3578 /* 3579 * Here's the tricky part. 3580 * 3581 * We need to move the pointer past the header page. 3582 * But we can only do that if a writer is not currently 3583 * moving it. The page before the header page has the 3584 * flag bit '1' set if it is pointing to the page we want. 3585 * but if the writer is in the process of moving it 3586 * than it will be '2' or already moved '0'. 3587 */ 3588 3589 ret = rb_head_page_replace(reader, cpu_buffer->reader_page); 3590 3591 /* 3592 * If we did not convert it, then we must try again. 3593 */ 3594 if (!ret) 3595 goto spin; 3596 3597 /* 3598 * Yeah! We succeeded in replacing the page. 3599 * 3600 * Now make the new head point back to the reader page. 3601 */ 3602 rb_list_head(reader->list.next)->prev = &cpu_buffer->reader_page->list; 3603 rb_inc_page(cpu_buffer, &cpu_buffer->head_page); 3604 3605 /* Finally update the reader page to the new head */ 3606 cpu_buffer->reader_page = reader; 3607 rb_reset_reader_page(cpu_buffer); 3608 3609 if (overwrite != cpu_buffer->last_overrun) { 3610 cpu_buffer->lost_events = overwrite - cpu_buffer->last_overrun; 3611 cpu_buffer->last_overrun = overwrite; 3612 } 3613 3614 goto again; 3615 3616 out: 3617 arch_spin_unlock(&cpu_buffer->lock); 3618 local_irq_restore(flags); 3619 3620 return reader; 3621 } 3622 3623 static void rb_advance_reader(struct ring_buffer_per_cpu *cpu_buffer) 3624 { 3625 struct ring_buffer_event *event; 3626 struct buffer_page *reader; 3627 unsigned length; 3628 3629 reader = rb_get_reader_page(cpu_buffer); 3630 3631 /* This function should not be called when buffer is empty */ 3632 if (RB_WARN_ON(cpu_buffer, !reader)) 3633 return; 3634 3635 event = rb_reader_event(cpu_buffer); 3636 3637 if (event->type_len <= RINGBUF_TYPE_DATA_TYPE_LEN_MAX) 3638 cpu_buffer->read++; 3639 3640 rb_update_read_stamp(cpu_buffer, event); 3641 3642 length = rb_event_length(event); 3643 cpu_buffer->reader_page->read += length; 3644 } 3645 3646 static void rb_advance_iter(struct ring_buffer_iter *iter) 3647 { 3648 struct ring_buffer_per_cpu *cpu_buffer; 3649 struct ring_buffer_event *event; 3650 unsigned length; 3651 3652 cpu_buffer = iter->cpu_buffer; 3653 3654 /* 3655 * Check if we are at the end of the buffer. 3656 */ 3657 if (iter->head >= rb_page_size(iter->head_page)) { 3658 /* discarded commits can make the page empty */ 3659 if (iter->head_page == cpu_buffer->commit_page) 3660 return; 3661 rb_inc_iter(iter); 3662 return; 3663 } 3664 3665 event = rb_iter_head_event(iter); 3666 3667 length = rb_event_length(event); 3668 3669 /* 3670 * This should not be called to advance the header if we are 3671 * at the tail of the buffer. 3672 */ 3673 if (RB_WARN_ON(cpu_buffer, 3674 (iter->head_page == cpu_buffer->commit_page) && 3675 (iter->head + length > rb_commit_index(cpu_buffer)))) 3676 return; 3677 3678 rb_update_iter_read_stamp(iter, event); 3679 3680 iter->head += length; 3681 3682 /* check for end of page padding */ 3683 if ((iter->head >= rb_page_size(iter->head_page)) && 3684 (iter->head_page != cpu_buffer->commit_page)) 3685 rb_inc_iter(iter); 3686 } 3687 3688 static int rb_lost_events(struct ring_buffer_per_cpu *cpu_buffer) 3689 { 3690 return cpu_buffer->lost_events; 3691 } 3692 3693 static struct ring_buffer_event * 3694 rb_buffer_peek(struct ring_buffer_per_cpu *cpu_buffer, u64 *ts, 3695 unsigned long *lost_events) 3696 { 3697 struct ring_buffer_event *event; 3698 struct buffer_page *reader; 3699 int nr_loops = 0; 3700 3701 again: 3702 /* 3703 * We repeat when a time extend is encountered. 3704 * Since the time extend is always attached to a data event, 3705 * we should never loop more than once. 3706 * (We never hit the following condition more than twice). 3707 */ 3708 if (RB_WARN_ON(cpu_buffer, ++nr_loops > 2)) 3709 return NULL; 3710 3711 reader = rb_get_reader_page(cpu_buffer); 3712 if (!reader) 3713 return NULL; 3714 3715 event = rb_reader_event(cpu_buffer); 3716 3717 switch (event->type_len) { 3718 case RINGBUF_TYPE_PADDING: 3719 if (rb_null_event(event)) 3720 RB_WARN_ON(cpu_buffer, 1); 3721 /* 3722 * Because the writer could be discarding every 3723 * event it creates (which would probably be bad) 3724 * if we were to go back to "again" then we may never 3725 * catch up, and will trigger the warn on, or lock 3726 * the box. Return the padding, and we will release 3727 * the current locks, and try again. 3728 */ 3729 return event; 3730 3731 case RINGBUF_TYPE_TIME_EXTEND: 3732 /* Internal data, OK to advance */ 3733 rb_advance_reader(cpu_buffer); 3734 goto again; 3735 3736 case RINGBUF_TYPE_TIME_STAMP: 3737 /* FIXME: not implemented */ 3738 rb_advance_reader(cpu_buffer); 3739 goto again; 3740 3741 case RINGBUF_TYPE_DATA: 3742 if (ts) { 3743 *ts = cpu_buffer->read_stamp + event->time_delta; 3744 ring_buffer_normalize_time_stamp(cpu_buffer->buffer, 3745 cpu_buffer->cpu, ts); 3746 } 3747 if (lost_events) 3748 *lost_events = rb_lost_events(cpu_buffer); 3749 return event; 3750 3751 default: 3752 BUG(); 3753 } 3754 3755 return NULL; 3756 } 3757 EXPORT_SYMBOL_GPL(ring_buffer_peek); 3758 3759 static struct ring_buffer_event * 3760 rb_iter_peek(struct ring_buffer_iter *iter, u64 *ts) 3761 { 3762 struct ring_buffer *buffer; 3763 struct ring_buffer_per_cpu *cpu_buffer; 3764 struct ring_buffer_event *event; 3765 int nr_loops = 0; 3766 3767 cpu_buffer = iter->cpu_buffer; 3768 buffer = cpu_buffer->buffer; 3769 3770 /* 3771 * Check if someone performed a consuming read to 3772 * the buffer. A consuming read invalidates the iterator 3773 * and we need to reset the iterator in this case. 3774 */ 3775 if (unlikely(iter->cache_read != cpu_buffer->read || 3776 iter->cache_reader_page != cpu_buffer->reader_page)) 3777 rb_iter_reset(iter); 3778 3779 again: 3780 if (ring_buffer_iter_empty(iter)) 3781 return NULL; 3782 3783 /* 3784 * We repeat when a time extend is encountered or we hit 3785 * the end of the page. Since the time extend is always attached 3786 * to a data event, we should never loop more than three times. 3787 * Once for going to next page, once on time extend, and 3788 * finally once to get the event. 3789 * (We never hit the following condition more than thrice). 3790 */ 3791 if (RB_WARN_ON(cpu_buffer, ++nr_loops > 3)) 3792 return NULL; 3793 3794 if (rb_per_cpu_empty(cpu_buffer)) 3795 return NULL; 3796 3797 if (iter->head >= rb_page_size(iter->head_page)) { 3798 rb_inc_iter(iter); 3799 goto again; 3800 } 3801 3802 event = rb_iter_head_event(iter); 3803 3804 switch (event->type_len) { 3805 case RINGBUF_TYPE_PADDING: 3806 if (rb_null_event(event)) { 3807 rb_inc_iter(iter); 3808 goto again; 3809 } 3810 rb_advance_iter(iter); 3811 return event; 3812 3813 case RINGBUF_TYPE_TIME_EXTEND: 3814 /* Internal data, OK to advance */ 3815 rb_advance_iter(iter); 3816 goto again; 3817 3818 case RINGBUF_TYPE_TIME_STAMP: 3819 /* FIXME: not implemented */ 3820 rb_advance_iter(iter); 3821 goto again; 3822 3823 case RINGBUF_TYPE_DATA: 3824 if (ts) { 3825 *ts = iter->read_stamp + event->time_delta; 3826 ring_buffer_normalize_time_stamp(buffer, 3827 cpu_buffer->cpu, ts); 3828 } 3829 return event; 3830 3831 default: 3832 BUG(); 3833 } 3834 3835 return NULL; 3836 } 3837 EXPORT_SYMBOL_GPL(ring_buffer_iter_peek); 3838 3839 static inline int rb_ok_to_lock(void) 3840 { 3841 /* 3842 * If an NMI die dumps out the content of the ring buffer 3843 * do not grab locks. We also permanently disable the ring 3844 * buffer too. A one time deal is all you get from reading 3845 * the ring buffer from an NMI. 3846 */ 3847 if (likely(!in_nmi())) 3848 return 1; 3849 3850 tracing_off_permanent(); 3851 return 0; 3852 } 3853 3854 /** 3855 * ring_buffer_peek - peek at the next event to be read 3856 * @buffer: The ring buffer to read 3857 * @cpu: The cpu to peak at 3858 * @ts: The timestamp counter of this event. 3859 * @lost_events: a variable to store if events were lost (may be NULL) 3860 * 3861 * This will return the event that will be read next, but does 3862 * not consume the data. 3863 */ 3864 struct ring_buffer_event * 3865 ring_buffer_peek(struct ring_buffer *buffer, int cpu, u64 *ts, 3866 unsigned long *lost_events) 3867 { 3868 struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu]; 3869 struct ring_buffer_event *event; 3870 unsigned long flags; 3871 int dolock; 3872 3873 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 3874 return NULL; 3875 3876 dolock = rb_ok_to_lock(); 3877 again: 3878 local_irq_save(flags); 3879 if (dolock) 3880 raw_spin_lock(&cpu_buffer->reader_lock); 3881 event = rb_buffer_peek(cpu_buffer, ts, lost_events); 3882 if (event && event->type_len == RINGBUF_TYPE_PADDING) 3883 rb_advance_reader(cpu_buffer); 3884 if (dolock) 3885 raw_spin_unlock(&cpu_buffer->reader_lock); 3886 local_irq_restore(flags); 3887 3888 if (event && event->type_len == RINGBUF_TYPE_PADDING) 3889 goto again; 3890 3891 return event; 3892 } 3893 3894 /** 3895 * ring_buffer_iter_peek - peek at the next event to be read 3896 * @iter: The ring buffer iterator 3897 * @ts: The timestamp counter of this event. 3898 * 3899 * This will return the event that will be read next, but does 3900 * not increment the iterator. 3901 */ 3902 struct ring_buffer_event * 3903 ring_buffer_iter_peek(struct ring_buffer_iter *iter, u64 *ts) 3904 { 3905 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer; 3906 struct ring_buffer_event *event; 3907 unsigned long flags; 3908 3909 again: 3910 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 3911 event = rb_iter_peek(iter, ts); 3912 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 3913 3914 if (event && event->type_len == RINGBUF_TYPE_PADDING) 3915 goto again; 3916 3917 return event; 3918 } 3919 3920 /** 3921 * ring_buffer_consume - return an event and consume it 3922 * @buffer: The ring buffer to get the next event from 3923 * @cpu: the cpu to read the buffer from 3924 * @ts: a variable to store the timestamp (may be NULL) 3925 * @lost_events: a variable to store if events were lost (may be NULL) 3926 * 3927 * Returns the next event in the ring buffer, and that event is consumed. 3928 * Meaning, that sequential reads will keep returning a different event, 3929 * and eventually empty the ring buffer if the producer is slower. 3930 */ 3931 struct ring_buffer_event * 3932 ring_buffer_consume(struct ring_buffer *buffer, int cpu, u64 *ts, 3933 unsigned long *lost_events) 3934 { 3935 struct ring_buffer_per_cpu *cpu_buffer; 3936 struct ring_buffer_event *event = NULL; 3937 unsigned long flags; 3938 int dolock; 3939 3940 dolock = rb_ok_to_lock(); 3941 3942 again: 3943 /* might be called in atomic */ 3944 preempt_disable(); 3945 3946 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 3947 goto out; 3948 3949 cpu_buffer = buffer->buffers[cpu]; 3950 local_irq_save(flags); 3951 if (dolock) 3952 raw_spin_lock(&cpu_buffer->reader_lock); 3953 3954 event = rb_buffer_peek(cpu_buffer, ts, lost_events); 3955 if (event) { 3956 cpu_buffer->lost_events = 0; 3957 rb_advance_reader(cpu_buffer); 3958 } 3959 3960 if (dolock) 3961 raw_spin_unlock(&cpu_buffer->reader_lock); 3962 local_irq_restore(flags); 3963 3964 out: 3965 preempt_enable(); 3966 3967 if (event && event->type_len == RINGBUF_TYPE_PADDING) 3968 goto again; 3969 3970 return event; 3971 } 3972 EXPORT_SYMBOL_GPL(ring_buffer_consume); 3973 3974 /** 3975 * ring_buffer_read_prepare - Prepare for a non consuming read of the buffer 3976 * @buffer: The ring buffer to read from 3977 * @cpu: The cpu buffer to iterate over 3978 * 3979 * This performs the initial preparations necessary to iterate 3980 * through the buffer. Memory is allocated, buffer recording 3981 * is disabled, and the iterator pointer is returned to the caller. 3982 * 3983 * Disabling buffer recordng prevents the reading from being 3984 * corrupted. This is not a consuming read, so a producer is not 3985 * expected. 3986 * 3987 * After a sequence of ring_buffer_read_prepare calls, the user is 3988 * expected to make at least one call to ring_buffer_read_prepare_sync. 3989 * Afterwards, ring_buffer_read_start is invoked to get things going 3990 * for real. 3991 * 3992 * This overall must be paired with ring_buffer_read_finish. 3993 */ 3994 struct ring_buffer_iter * 3995 ring_buffer_read_prepare(struct ring_buffer *buffer, int cpu) 3996 { 3997 struct ring_buffer_per_cpu *cpu_buffer; 3998 struct ring_buffer_iter *iter; 3999 4000 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 4001 return NULL; 4002 4003 iter = kmalloc(sizeof(*iter), GFP_KERNEL); 4004 if (!iter) 4005 return NULL; 4006 4007 cpu_buffer = buffer->buffers[cpu]; 4008 4009 iter->cpu_buffer = cpu_buffer; 4010 4011 atomic_inc(&buffer->resize_disabled); 4012 atomic_inc(&cpu_buffer->record_disabled); 4013 4014 return iter; 4015 } 4016 EXPORT_SYMBOL_GPL(ring_buffer_read_prepare); 4017 4018 /** 4019 * ring_buffer_read_prepare_sync - Synchronize a set of prepare calls 4020 * 4021 * All previously invoked ring_buffer_read_prepare calls to prepare 4022 * iterators will be synchronized. Afterwards, read_buffer_read_start 4023 * calls on those iterators are allowed. 4024 */ 4025 void 4026 ring_buffer_read_prepare_sync(void) 4027 { 4028 synchronize_sched(); 4029 } 4030 EXPORT_SYMBOL_GPL(ring_buffer_read_prepare_sync); 4031 4032 /** 4033 * ring_buffer_read_start - start a non consuming read of the buffer 4034 * @iter: The iterator returned by ring_buffer_read_prepare 4035 * 4036 * This finalizes the startup of an iteration through the buffer. 4037 * The iterator comes from a call to ring_buffer_read_prepare and 4038 * an intervening ring_buffer_read_prepare_sync must have been 4039 * performed. 4040 * 4041 * Must be paired with ring_buffer_read_finish. 4042 */ 4043 void 4044 ring_buffer_read_start(struct ring_buffer_iter *iter) 4045 { 4046 struct ring_buffer_per_cpu *cpu_buffer; 4047 unsigned long flags; 4048 4049 if (!iter) 4050 return; 4051 4052 cpu_buffer = iter->cpu_buffer; 4053 4054 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 4055 arch_spin_lock(&cpu_buffer->lock); 4056 rb_iter_reset(iter); 4057 arch_spin_unlock(&cpu_buffer->lock); 4058 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 4059 } 4060 EXPORT_SYMBOL_GPL(ring_buffer_read_start); 4061 4062 /** 4063 * ring_buffer_read_finish - finish reading the iterator of the buffer 4064 * @iter: The iterator retrieved by ring_buffer_start 4065 * 4066 * This re-enables the recording to the buffer, and frees the 4067 * iterator. 4068 */ 4069 void 4070 ring_buffer_read_finish(struct ring_buffer_iter *iter) 4071 { 4072 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer; 4073 unsigned long flags; 4074 4075 /* 4076 * Ring buffer is disabled from recording, here's a good place 4077 * to check the integrity of the ring buffer. 4078 * Must prevent readers from trying to read, as the check 4079 * clears the HEAD page and readers require it. 4080 */ 4081 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 4082 rb_check_pages(cpu_buffer); 4083 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 4084 4085 atomic_dec(&cpu_buffer->record_disabled); 4086 atomic_dec(&cpu_buffer->buffer->resize_disabled); 4087 kfree(iter); 4088 } 4089 EXPORT_SYMBOL_GPL(ring_buffer_read_finish); 4090 4091 /** 4092 * ring_buffer_read - read the next item in the ring buffer by the iterator 4093 * @iter: The ring buffer iterator 4094 * @ts: The time stamp of the event read. 4095 * 4096 * This reads the next event in the ring buffer and increments the iterator. 4097 */ 4098 struct ring_buffer_event * 4099 ring_buffer_read(struct ring_buffer_iter *iter, u64 *ts) 4100 { 4101 struct ring_buffer_event *event; 4102 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer; 4103 unsigned long flags; 4104 4105 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 4106 again: 4107 event = rb_iter_peek(iter, ts); 4108 if (!event) 4109 goto out; 4110 4111 if (event->type_len == RINGBUF_TYPE_PADDING) 4112 goto again; 4113 4114 rb_advance_iter(iter); 4115 out: 4116 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 4117 4118 return event; 4119 } 4120 EXPORT_SYMBOL_GPL(ring_buffer_read); 4121 4122 /** 4123 * ring_buffer_size - return the size of the ring buffer (in bytes) 4124 * @buffer: The ring buffer. 4125 */ 4126 unsigned long ring_buffer_size(struct ring_buffer *buffer, int cpu) 4127 { 4128 /* 4129 * Earlier, this method returned 4130 * BUF_PAGE_SIZE * buffer->nr_pages 4131 * Since the nr_pages field is now removed, we have converted this to 4132 * return the per cpu buffer value. 4133 */ 4134 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 4135 return 0; 4136 4137 return BUF_PAGE_SIZE * buffer->buffers[cpu]->nr_pages; 4138 } 4139 EXPORT_SYMBOL_GPL(ring_buffer_size); 4140 4141 static void 4142 rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer) 4143 { 4144 rb_head_page_deactivate(cpu_buffer); 4145 4146 cpu_buffer->head_page 4147 = list_entry(cpu_buffer->pages, struct buffer_page, list); 4148 local_set(&cpu_buffer->head_page->write, 0); 4149 local_set(&cpu_buffer->head_page->entries, 0); 4150 local_set(&cpu_buffer->head_page->page->commit, 0); 4151 4152 cpu_buffer->head_page->read = 0; 4153 4154 cpu_buffer->tail_page = cpu_buffer->head_page; 4155 cpu_buffer->commit_page = cpu_buffer->head_page; 4156 4157 INIT_LIST_HEAD(&cpu_buffer->reader_page->list); 4158 INIT_LIST_HEAD(&cpu_buffer->new_pages); 4159 local_set(&cpu_buffer->reader_page->write, 0); 4160 local_set(&cpu_buffer->reader_page->entries, 0); 4161 local_set(&cpu_buffer->reader_page->page->commit, 0); 4162 cpu_buffer->reader_page->read = 0; 4163 4164 local_set(&cpu_buffer->entries_bytes, 0); 4165 local_set(&cpu_buffer->overrun, 0); 4166 local_set(&cpu_buffer->commit_overrun, 0); 4167 local_set(&cpu_buffer->dropped_events, 0); 4168 local_set(&cpu_buffer->entries, 0); 4169 local_set(&cpu_buffer->committing, 0); 4170 local_set(&cpu_buffer->commits, 0); 4171 cpu_buffer->read = 0; 4172 cpu_buffer->read_bytes = 0; 4173 4174 cpu_buffer->write_stamp = 0; 4175 cpu_buffer->read_stamp = 0; 4176 4177 cpu_buffer->lost_events = 0; 4178 cpu_buffer->last_overrun = 0; 4179 4180 rb_head_page_activate(cpu_buffer); 4181 } 4182 4183 /** 4184 * ring_buffer_reset_cpu - reset a ring buffer per CPU buffer 4185 * @buffer: The ring buffer to reset a per cpu buffer of 4186 * @cpu: The CPU buffer to be reset 4187 */ 4188 void ring_buffer_reset_cpu(struct ring_buffer *buffer, int cpu) 4189 { 4190 struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu]; 4191 unsigned long flags; 4192 4193 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 4194 return; 4195 4196 atomic_inc(&buffer->resize_disabled); 4197 atomic_inc(&cpu_buffer->record_disabled); 4198 4199 /* Make sure all commits have finished */ 4200 synchronize_sched(); 4201 4202 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 4203 4204 if (RB_WARN_ON(cpu_buffer, local_read(&cpu_buffer->committing))) 4205 goto out; 4206 4207 arch_spin_lock(&cpu_buffer->lock); 4208 4209 rb_reset_cpu(cpu_buffer); 4210 4211 arch_spin_unlock(&cpu_buffer->lock); 4212 4213 out: 4214 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 4215 4216 atomic_dec(&cpu_buffer->record_disabled); 4217 atomic_dec(&buffer->resize_disabled); 4218 } 4219 EXPORT_SYMBOL_GPL(ring_buffer_reset_cpu); 4220 4221 /** 4222 * ring_buffer_reset - reset a ring buffer 4223 * @buffer: The ring buffer to reset all cpu buffers 4224 */ 4225 void ring_buffer_reset(struct ring_buffer *buffer) 4226 { 4227 int cpu; 4228 4229 for_each_buffer_cpu(buffer, cpu) 4230 ring_buffer_reset_cpu(buffer, cpu); 4231 } 4232 EXPORT_SYMBOL_GPL(ring_buffer_reset); 4233 4234 /** 4235 * rind_buffer_empty - is the ring buffer empty? 4236 * @buffer: The ring buffer to test 4237 */ 4238 int ring_buffer_empty(struct ring_buffer *buffer) 4239 { 4240 struct ring_buffer_per_cpu *cpu_buffer; 4241 unsigned long flags; 4242 int dolock; 4243 int cpu; 4244 int ret; 4245 4246 dolock = rb_ok_to_lock(); 4247 4248 /* yes this is racy, but if you don't like the race, lock the buffer */ 4249 for_each_buffer_cpu(buffer, cpu) { 4250 cpu_buffer = buffer->buffers[cpu]; 4251 local_irq_save(flags); 4252 if (dolock) 4253 raw_spin_lock(&cpu_buffer->reader_lock); 4254 ret = rb_per_cpu_empty(cpu_buffer); 4255 if (dolock) 4256 raw_spin_unlock(&cpu_buffer->reader_lock); 4257 local_irq_restore(flags); 4258 4259 if (!ret) 4260 return 0; 4261 } 4262 4263 return 1; 4264 } 4265 EXPORT_SYMBOL_GPL(ring_buffer_empty); 4266 4267 /** 4268 * ring_buffer_empty_cpu - is a cpu buffer of a ring buffer empty? 4269 * @buffer: The ring buffer 4270 * @cpu: The CPU buffer to test 4271 */ 4272 int ring_buffer_empty_cpu(struct ring_buffer *buffer, int cpu) 4273 { 4274 struct ring_buffer_per_cpu *cpu_buffer; 4275 unsigned long flags; 4276 int dolock; 4277 int ret; 4278 4279 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 4280 return 1; 4281 4282 dolock = rb_ok_to_lock(); 4283 4284 cpu_buffer = buffer->buffers[cpu]; 4285 local_irq_save(flags); 4286 if (dolock) 4287 raw_spin_lock(&cpu_buffer->reader_lock); 4288 ret = rb_per_cpu_empty(cpu_buffer); 4289 if (dolock) 4290 raw_spin_unlock(&cpu_buffer->reader_lock); 4291 local_irq_restore(flags); 4292 4293 return ret; 4294 } 4295 EXPORT_SYMBOL_GPL(ring_buffer_empty_cpu); 4296 4297 #ifdef CONFIG_RING_BUFFER_ALLOW_SWAP 4298 /** 4299 * ring_buffer_swap_cpu - swap a CPU buffer between two ring buffers 4300 * @buffer_a: One buffer to swap with 4301 * @buffer_b: The other buffer to swap with 4302 * 4303 * This function is useful for tracers that want to take a "snapshot" 4304 * of a CPU buffer and has another back up buffer lying around. 4305 * it is expected that the tracer handles the cpu buffer not being 4306 * used at the moment. 4307 */ 4308 int ring_buffer_swap_cpu(struct ring_buffer *buffer_a, 4309 struct ring_buffer *buffer_b, int cpu) 4310 { 4311 struct ring_buffer_per_cpu *cpu_buffer_a; 4312 struct ring_buffer_per_cpu *cpu_buffer_b; 4313 int ret = -EINVAL; 4314 4315 if (!cpumask_test_cpu(cpu, buffer_a->cpumask) || 4316 !cpumask_test_cpu(cpu, buffer_b->cpumask)) 4317 goto out; 4318 4319 cpu_buffer_a = buffer_a->buffers[cpu]; 4320 cpu_buffer_b = buffer_b->buffers[cpu]; 4321 4322 /* At least make sure the two buffers are somewhat the same */ 4323 if (cpu_buffer_a->nr_pages != cpu_buffer_b->nr_pages) 4324 goto out; 4325 4326 ret = -EAGAIN; 4327 4328 if (ring_buffer_flags != RB_BUFFERS_ON) 4329 goto out; 4330 4331 if (atomic_read(&buffer_a->record_disabled)) 4332 goto out; 4333 4334 if (atomic_read(&buffer_b->record_disabled)) 4335 goto out; 4336 4337 if (atomic_read(&cpu_buffer_a->record_disabled)) 4338 goto out; 4339 4340 if (atomic_read(&cpu_buffer_b->record_disabled)) 4341 goto out; 4342 4343 /* 4344 * We can't do a synchronize_sched here because this 4345 * function can be called in atomic context. 4346 * Normally this will be called from the same CPU as cpu. 4347 * If not it's up to the caller to protect this. 4348 */ 4349 atomic_inc(&cpu_buffer_a->record_disabled); 4350 atomic_inc(&cpu_buffer_b->record_disabled); 4351 4352 ret = -EBUSY; 4353 if (local_read(&cpu_buffer_a->committing)) 4354 goto out_dec; 4355 if (local_read(&cpu_buffer_b->committing)) 4356 goto out_dec; 4357 4358 buffer_a->buffers[cpu] = cpu_buffer_b; 4359 buffer_b->buffers[cpu] = cpu_buffer_a; 4360 4361 cpu_buffer_b->buffer = buffer_a; 4362 cpu_buffer_a->buffer = buffer_b; 4363 4364 ret = 0; 4365 4366 out_dec: 4367 atomic_dec(&cpu_buffer_a->record_disabled); 4368 atomic_dec(&cpu_buffer_b->record_disabled); 4369 out: 4370 return ret; 4371 } 4372 EXPORT_SYMBOL_GPL(ring_buffer_swap_cpu); 4373 #endif /* CONFIG_RING_BUFFER_ALLOW_SWAP */ 4374 4375 /** 4376 * ring_buffer_alloc_read_page - allocate a page to read from buffer 4377 * @buffer: the buffer to allocate for. 4378 * @cpu: the cpu buffer to allocate. 4379 * 4380 * This function is used in conjunction with ring_buffer_read_page. 4381 * When reading a full page from the ring buffer, these functions 4382 * can be used to speed up the process. The calling function should 4383 * allocate a few pages first with this function. Then when it 4384 * needs to get pages from the ring buffer, it passes the result 4385 * of this function into ring_buffer_read_page, which will swap 4386 * the page that was allocated, with the read page of the buffer. 4387 * 4388 * Returns: 4389 * The page allocated, or NULL on error. 4390 */ 4391 void *ring_buffer_alloc_read_page(struct ring_buffer *buffer, int cpu) 4392 { 4393 struct buffer_data_page *bpage; 4394 struct page *page; 4395 4396 page = alloc_pages_node(cpu_to_node(cpu), 4397 GFP_KERNEL | __GFP_NORETRY, 0); 4398 if (!page) 4399 return NULL; 4400 4401 bpage = page_address(page); 4402 4403 rb_init_page(bpage); 4404 4405 return bpage; 4406 } 4407 EXPORT_SYMBOL_GPL(ring_buffer_alloc_read_page); 4408 4409 /** 4410 * ring_buffer_free_read_page - free an allocated read page 4411 * @buffer: the buffer the page was allocate for 4412 * @data: the page to free 4413 * 4414 * Free a page allocated from ring_buffer_alloc_read_page. 4415 */ 4416 void ring_buffer_free_read_page(struct ring_buffer *buffer, void *data) 4417 { 4418 free_page((unsigned long)data); 4419 } 4420 EXPORT_SYMBOL_GPL(ring_buffer_free_read_page); 4421 4422 /** 4423 * ring_buffer_read_page - extract a page from the ring buffer 4424 * @buffer: buffer to extract from 4425 * @data_page: the page to use allocated from ring_buffer_alloc_read_page 4426 * @len: amount to extract 4427 * @cpu: the cpu of the buffer to extract 4428 * @full: should the extraction only happen when the page is full. 4429 * 4430 * This function will pull out a page from the ring buffer and consume it. 4431 * @data_page must be the address of the variable that was returned 4432 * from ring_buffer_alloc_read_page. This is because the page might be used 4433 * to swap with a page in the ring buffer. 4434 * 4435 * for example: 4436 * rpage = ring_buffer_alloc_read_page(buffer, cpu); 4437 * if (!rpage) 4438 * return error; 4439 * ret = ring_buffer_read_page(buffer, &rpage, len, cpu, 0); 4440 * if (ret >= 0) 4441 * process_page(rpage, ret); 4442 * 4443 * When @full is set, the function will not return true unless 4444 * the writer is off the reader page. 4445 * 4446 * Note: it is up to the calling functions to handle sleeps and wakeups. 4447 * The ring buffer can be used anywhere in the kernel and can not 4448 * blindly call wake_up. The layer that uses the ring buffer must be 4449 * responsible for that. 4450 * 4451 * Returns: 4452 * >=0 if data has been transferred, returns the offset of consumed data. 4453 * <0 if no data has been transferred. 4454 */ 4455 int ring_buffer_read_page(struct ring_buffer *buffer, 4456 void **data_page, size_t len, int cpu, int full) 4457 { 4458 struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu]; 4459 struct ring_buffer_event *event; 4460 struct buffer_data_page *bpage; 4461 struct buffer_page *reader; 4462 unsigned long missed_events; 4463 unsigned long flags; 4464 unsigned int commit; 4465 unsigned int read; 4466 u64 save_timestamp; 4467 int ret = -1; 4468 4469 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 4470 goto out; 4471 4472 /* 4473 * If len is not big enough to hold the page header, then 4474 * we can not copy anything. 4475 */ 4476 if (len <= BUF_PAGE_HDR_SIZE) 4477 goto out; 4478 4479 len -= BUF_PAGE_HDR_SIZE; 4480 4481 if (!data_page) 4482 goto out; 4483 4484 bpage = *data_page; 4485 if (!bpage) 4486 goto out; 4487 4488 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 4489 4490 reader = rb_get_reader_page(cpu_buffer); 4491 if (!reader) 4492 goto out_unlock; 4493 4494 event = rb_reader_event(cpu_buffer); 4495 4496 read = reader->read; 4497 commit = rb_page_commit(reader); 4498 4499 /* Check if any events were dropped */ 4500 missed_events = cpu_buffer->lost_events; 4501 4502 /* 4503 * If this page has been partially read or 4504 * if len is not big enough to read the rest of the page or 4505 * a writer is still on the page, then 4506 * we must copy the data from the page to the buffer. 4507 * Otherwise, we can simply swap the page with the one passed in. 4508 */ 4509 if (read || (len < (commit - read)) || 4510 cpu_buffer->reader_page == cpu_buffer->commit_page) { 4511 struct buffer_data_page *rpage = cpu_buffer->reader_page->page; 4512 unsigned int rpos = read; 4513 unsigned int pos = 0; 4514 unsigned int size; 4515 4516 if (full) 4517 goto out_unlock; 4518 4519 if (len > (commit - read)) 4520 len = (commit - read); 4521 4522 /* Always keep the time extend and data together */ 4523 size = rb_event_ts_length(event); 4524 4525 if (len < size) 4526 goto out_unlock; 4527 4528 /* save the current timestamp, since the user will need it */ 4529 save_timestamp = cpu_buffer->read_stamp; 4530 4531 /* Need to copy one event at a time */ 4532 do { 4533 /* We need the size of one event, because 4534 * rb_advance_reader only advances by one event, 4535 * whereas rb_event_ts_length may include the size of 4536 * one or two events. 4537 * We have already ensured there's enough space if this 4538 * is a time extend. */ 4539 size = rb_event_length(event); 4540 memcpy(bpage->data + pos, rpage->data + rpos, size); 4541 4542 len -= size; 4543 4544 rb_advance_reader(cpu_buffer); 4545 rpos = reader->read; 4546 pos += size; 4547 4548 if (rpos >= commit) 4549 break; 4550 4551 event = rb_reader_event(cpu_buffer); 4552 /* Always keep the time extend and data together */ 4553 size = rb_event_ts_length(event); 4554 } while (len >= size); 4555 4556 /* update bpage */ 4557 local_set(&bpage->commit, pos); 4558 bpage->time_stamp = save_timestamp; 4559 4560 /* we copied everything to the beginning */ 4561 read = 0; 4562 } else { 4563 /* update the entry counter */ 4564 cpu_buffer->read += rb_page_entries(reader); 4565 cpu_buffer->read_bytes += BUF_PAGE_SIZE; 4566 4567 /* swap the pages */ 4568 rb_init_page(bpage); 4569 bpage = reader->page; 4570 reader->page = *data_page; 4571 local_set(&reader->write, 0); 4572 local_set(&reader->entries, 0); 4573 reader->read = 0; 4574 *data_page = bpage; 4575 4576 /* 4577 * Use the real_end for the data size, 4578 * This gives us a chance to store the lost events 4579 * on the page. 4580 */ 4581 if (reader->real_end) 4582 local_set(&bpage->commit, reader->real_end); 4583 } 4584 ret = read; 4585 4586 cpu_buffer->lost_events = 0; 4587 4588 commit = local_read(&bpage->commit); 4589 /* 4590 * Set a flag in the commit field if we lost events 4591 */ 4592 if (missed_events) { 4593 /* If there is room at the end of the page to save the 4594 * missed events, then record it there. 4595 */ 4596 if (BUF_PAGE_SIZE - commit >= sizeof(missed_events)) { 4597 memcpy(&bpage->data[commit], &missed_events, 4598 sizeof(missed_events)); 4599 local_add(RB_MISSED_STORED, &bpage->commit); 4600 commit += sizeof(missed_events); 4601 } 4602 local_add(RB_MISSED_EVENTS, &bpage->commit); 4603 } 4604 4605 /* 4606 * This page may be off to user land. Zero it out here. 4607 */ 4608 if (commit < BUF_PAGE_SIZE) 4609 memset(&bpage->data[commit], 0, BUF_PAGE_SIZE - commit); 4610 4611 out_unlock: 4612 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 4613 4614 out: 4615 return ret; 4616 } 4617 EXPORT_SYMBOL_GPL(ring_buffer_read_page); 4618 4619 #ifdef CONFIG_HOTPLUG_CPU 4620 static int rb_cpu_notify(struct notifier_block *self, 4621 unsigned long action, void *hcpu) 4622 { 4623 struct ring_buffer *buffer = 4624 container_of(self, struct ring_buffer, cpu_notify); 4625 long cpu = (long)hcpu; 4626 int cpu_i, nr_pages_same; 4627 unsigned int nr_pages; 4628 4629 switch (action) { 4630 case CPU_UP_PREPARE: 4631 case CPU_UP_PREPARE_FROZEN: 4632 if (cpumask_test_cpu(cpu, buffer->cpumask)) 4633 return NOTIFY_OK; 4634 4635 nr_pages = 0; 4636 nr_pages_same = 1; 4637 /* check if all cpu sizes are same */ 4638 for_each_buffer_cpu(buffer, cpu_i) { 4639 /* fill in the size from first enabled cpu */ 4640 if (nr_pages == 0) 4641 nr_pages = buffer->buffers[cpu_i]->nr_pages; 4642 if (nr_pages != buffer->buffers[cpu_i]->nr_pages) { 4643 nr_pages_same = 0; 4644 break; 4645 } 4646 } 4647 /* allocate minimum pages, user can later expand it */ 4648 if (!nr_pages_same) 4649 nr_pages = 2; 4650 buffer->buffers[cpu] = 4651 rb_allocate_cpu_buffer(buffer, nr_pages, cpu); 4652 if (!buffer->buffers[cpu]) { 4653 WARN(1, "failed to allocate ring buffer on CPU %ld\n", 4654 cpu); 4655 return NOTIFY_OK; 4656 } 4657 smp_wmb(); 4658 cpumask_set_cpu(cpu, buffer->cpumask); 4659 break; 4660 case CPU_DOWN_PREPARE: 4661 case CPU_DOWN_PREPARE_FROZEN: 4662 /* 4663 * Do nothing. 4664 * If we were to free the buffer, then the user would 4665 * lose any trace that was in the buffer. 4666 */ 4667 break; 4668 default: 4669 break; 4670 } 4671 return NOTIFY_OK; 4672 } 4673 #endif 4674 4675 #ifdef CONFIG_RING_BUFFER_STARTUP_TEST 4676 /* 4677 * This is a basic integrity check of the ring buffer. 4678 * Late in the boot cycle this test will run when configured in. 4679 * It will kick off a thread per CPU that will go into a loop 4680 * writing to the per cpu ring buffer various sizes of data. 4681 * Some of the data will be large items, some small. 4682 * 4683 * Another thread is created that goes into a spin, sending out 4684 * IPIs to the other CPUs to also write into the ring buffer. 4685 * this is to test the nesting ability of the buffer. 4686 * 4687 * Basic stats are recorded and reported. If something in the 4688 * ring buffer should happen that's not expected, a big warning 4689 * is displayed and all ring buffers are disabled. 4690 */ 4691 static struct task_struct *rb_threads[NR_CPUS] __initdata; 4692 4693 struct rb_test_data { 4694 struct ring_buffer *buffer; 4695 unsigned long events; 4696 unsigned long bytes_written; 4697 unsigned long bytes_alloc; 4698 unsigned long bytes_dropped; 4699 unsigned long events_nested; 4700 unsigned long bytes_written_nested; 4701 unsigned long bytes_alloc_nested; 4702 unsigned long bytes_dropped_nested; 4703 int min_size_nested; 4704 int max_size_nested; 4705 int max_size; 4706 int min_size; 4707 int cpu; 4708 int cnt; 4709 }; 4710 4711 static struct rb_test_data rb_data[NR_CPUS] __initdata; 4712 4713 /* 1 meg per cpu */ 4714 #define RB_TEST_BUFFER_SIZE 1048576 4715 4716 static char rb_string[] __initdata = 4717 "abcdefghijklmnopqrstuvwxyz1234567890!@#$%^&*()?+\\" 4718 "?+|:';\",.<>/?abcdefghijklmnopqrstuvwxyz1234567890" 4719 "!@#$%^&*()?+\\?+|:';\",.<>/?abcdefghijklmnopqrstuv"; 4720 4721 static bool rb_test_started __initdata; 4722 4723 struct rb_item { 4724 int size; 4725 char str[]; 4726 }; 4727 4728 static __init int rb_write_something(struct rb_test_data *data, bool nested) 4729 { 4730 struct ring_buffer_event *event; 4731 struct rb_item *item; 4732 bool started; 4733 int event_len; 4734 int size; 4735 int len; 4736 int cnt; 4737 4738 /* Have nested writes different that what is written */ 4739 cnt = data->cnt + (nested ? 27 : 0); 4740 4741 /* Multiply cnt by ~e, to make some unique increment */ 4742 size = (data->cnt * 68 / 25) % (sizeof(rb_string) - 1); 4743 4744 len = size + sizeof(struct rb_item); 4745 4746 started = rb_test_started; 4747 /* read rb_test_started before checking buffer enabled */ 4748 smp_rmb(); 4749 4750 event = ring_buffer_lock_reserve(data->buffer, len); 4751 if (!event) { 4752 /* Ignore dropped events before test starts. */ 4753 if (started) { 4754 if (nested) 4755 data->bytes_dropped += len; 4756 else 4757 data->bytes_dropped_nested += len; 4758 } 4759 return len; 4760 } 4761 4762 event_len = ring_buffer_event_length(event); 4763 4764 if (RB_WARN_ON(data->buffer, event_len < len)) 4765 goto out; 4766 4767 item = ring_buffer_event_data(event); 4768 item->size = size; 4769 memcpy(item->str, rb_string, size); 4770 4771 if (nested) { 4772 data->bytes_alloc_nested += event_len; 4773 data->bytes_written_nested += len; 4774 data->events_nested++; 4775 if (!data->min_size_nested || len < data->min_size_nested) 4776 data->min_size_nested = len; 4777 if (len > data->max_size_nested) 4778 data->max_size_nested = len; 4779 } else { 4780 data->bytes_alloc += event_len; 4781 data->bytes_written += len; 4782 data->events++; 4783 if (!data->min_size || len < data->min_size) 4784 data->max_size = len; 4785 if (len > data->max_size) 4786 data->max_size = len; 4787 } 4788 4789 out: 4790 ring_buffer_unlock_commit(data->buffer, event); 4791 4792 return 0; 4793 } 4794 4795 static __init int rb_test(void *arg) 4796 { 4797 struct rb_test_data *data = arg; 4798 4799 while (!kthread_should_stop()) { 4800 rb_write_something(data, false); 4801 data->cnt++; 4802 4803 set_current_state(TASK_INTERRUPTIBLE); 4804 /* Now sleep between a min of 100-300us and a max of 1ms */ 4805 usleep_range(((data->cnt % 3) + 1) * 100, 1000); 4806 } 4807 4808 return 0; 4809 } 4810 4811 static __init void rb_ipi(void *ignore) 4812 { 4813 struct rb_test_data *data; 4814 int cpu = smp_processor_id(); 4815 4816 data = &rb_data[cpu]; 4817 rb_write_something(data, true); 4818 } 4819 4820 static __init int rb_hammer_test(void *arg) 4821 { 4822 while (!kthread_should_stop()) { 4823 4824 /* Send an IPI to all cpus to write data! */ 4825 smp_call_function(rb_ipi, NULL, 1); 4826 /* No sleep, but for non preempt, let others run */ 4827 schedule(); 4828 } 4829 4830 return 0; 4831 } 4832 4833 static __init int test_ringbuffer(void) 4834 { 4835 struct task_struct *rb_hammer; 4836 struct ring_buffer *buffer; 4837 int cpu; 4838 int ret = 0; 4839 4840 pr_info("Running ring buffer tests...\n"); 4841 4842 buffer = ring_buffer_alloc(RB_TEST_BUFFER_SIZE, RB_FL_OVERWRITE); 4843 if (WARN_ON(!buffer)) 4844 return 0; 4845 4846 /* Disable buffer so that threads can't write to it yet */ 4847 ring_buffer_record_off(buffer); 4848 4849 for_each_online_cpu(cpu) { 4850 rb_data[cpu].buffer = buffer; 4851 rb_data[cpu].cpu = cpu; 4852 rb_data[cpu].cnt = cpu; 4853 rb_threads[cpu] = kthread_create(rb_test, &rb_data[cpu], 4854 "rbtester/%d", cpu); 4855 if (WARN_ON(!rb_threads[cpu])) { 4856 pr_cont("FAILED\n"); 4857 ret = -1; 4858 goto out_free; 4859 } 4860 4861 kthread_bind(rb_threads[cpu], cpu); 4862 wake_up_process(rb_threads[cpu]); 4863 } 4864 4865 /* Now create the rb hammer! */ 4866 rb_hammer = kthread_run(rb_hammer_test, NULL, "rbhammer"); 4867 if (WARN_ON(!rb_hammer)) { 4868 pr_cont("FAILED\n"); 4869 ret = -1; 4870 goto out_free; 4871 } 4872 4873 ring_buffer_record_on(buffer); 4874 /* 4875 * Show buffer is enabled before setting rb_test_started. 4876 * Yes there's a small race window where events could be 4877 * dropped and the thread wont catch it. But when a ring 4878 * buffer gets enabled, there will always be some kind of 4879 * delay before other CPUs see it. Thus, we don't care about 4880 * those dropped events. We care about events dropped after 4881 * the threads see that the buffer is active. 4882 */ 4883 smp_wmb(); 4884 rb_test_started = true; 4885 4886 set_current_state(TASK_INTERRUPTIBLE); 4887 /* Just run for 10 seconds */; 4888 schedule_timeout(10 * HZ); 4889 4890 kthread_stop(rb_hammer); 4891 4892 out_free: 4893 for_each_online_cpu(cpu) { 4894 if (!rb_threads[cpu]) 4895 break; 4896 kthread_stop(rb_threads[cpu]); 4897 } 4898 if (ret) { 4899 ring_buffer_free(buffer); 4900 return ret; 4901 } 4902 4903 /* Report! */ 4904 pr_info("finished\n"); 4905 for_each_online_cpu(cpu) { 4906 struct ring_buffer_event *event; 4907 struct rb_test_data *data = &rb_data[cpu]; 4908 struct rb_item *item; 4909 unsigned long total_events; 4910 unsigned long total_dropped; 4911 unsigned long total_written; 4912 unsigned long total_alloc; 4913 unsigned long total_read = 0; 4914 unsigned long total_size = 0; 4915 unsigned long total_len = 0; 4916 unsigned long total_lost = 0; 4917 unsigned long lost; 4918 int big_event_size; 4919 int small_event_size; 4920 4921 ret = -1; 4922 4923 total_events = data->events + data->events_nested; 4924 total_written = data->bytes_written + data->bytes_written_nested; 4925 total_alloc = data->bytes_alloc + data->bytes_alloc_nested; 4926 total_dropped = data->bytes_dropped + data->bytes_dropped_nested; 4927 4928 big_event_size = data->max_size + data->max_size_nested; 4929 small_event_size = data->min_size + data->min_size_nested; 4930 4931 pr_info("CPU %d:\n", cpu); 4932 pr_info(" events: %ld\n", total_events); 4933 pr_info(" dropped bytes: %ld\n", total_dropped); 4934 pr_info(" alloced bytes: %ld\n", total_alloc); 4935 pr_info(" written bytes: %ld\n", total_written); 4936 pr_info(" biggest event: %d\n", big_event_size); 4937 pr_info(" smallest event: %d\n", small_event_size); 4938 4939 if (RB_WARN_ON(buffer, total_dropped)) 4940 break; 4941 4942 ret = 0; 4943 4944 while ((event = ring_buffer_consume(buffer, cpu, NULL, &lost))) { 4945 total_lost += lost; 4946 item = ring_buffer_event_data(event); 4947 total_len += ring_buffer_event_length(event); 4948 total_size += item->size + sizeof(struct rb_item); 4949 if (memcmp(&item->str[0], rb_string, item->size) != 0) { 4950 pr_info("FAILED!\n"); 4951 pr_info("buffer had: %.*s\n", item->size, item->str); 4952 pr_info("expected: %.*s\n", item->size, rb_string); 4953 RB_WARN_ON(buffer, 1); 4954 ret = -1; 4955 break; 4956 } 4957 total_read++; 4958 } 4959 if (ret) 4960 break; 4961 4962 ret = -1; 4963 4964 pr_info(" read events: %ld\n", total_read); 4965 pr_info(" lost events: %ld\n", total_lost); 4966 pr_info(" total events: %ld\n", total_lost + total_read); 4967 pr_info(" recorded len bytes: %ld\n", total_len); 4968 pr_info(" recorded size bytes: %ld\n", total_size); 4969 if (total_lost) 4970 pr_info(" With dropped events, record len and size may not match\n" 4971 " alloced and written from above\n"); 4972 if (!total_lost) { 4973 if (RB_WARN_ON(buffer, total_len != total_alloc || 4974 total_size != total_written)) 4975 break; 4976 } 4977 if (RB_WARN_ON(buffer, total_lost + total_read != total_events)) 4978 break; 4979 4980 ret = 0; 4981 } 4982 if (!ret) 4983 pr_info("Ring buffer PASSED!\n"); 4984 4985 ring_buffer_free(buffer); 4986 return 0; 4987 } 4988 4989 late_initcall(test_ringbuffer); 4990 #endif /* CONFIG_RING_BUFFER_STARTUP_TEST */ 4991