FILE
ring_buffer.c
File preview with syntax highlighting.
linux-6.19.14/kernel/trace/ring_buffer.c
1// SPDX-License-Identifier: GPL-2.0 2/* 3 * Generic ring buffer 4 * 5 * Copyright (C) 2008 Steven Rostedt <srostedt@redhat.com> 6 */ 7#include <linux/trace_recursion.h> 8#include <linux/trace_events.h> 9#include <linux/ring_buffer.h> 10#include <linux/trace_clock.h> 11#include <linux/sched/clock.h> 12#include <linux/cacheflush.h> 13#include <linux/trace_seq.h> 14#include <linux/spinlock.h> 15#include <linux/irq_work.h> 16#include <linux/security.h> 17#include <linux/uaccess.h> 18#include <linux/hardirq.h> 19#include <linux/kthread.h> /* for self test */ 20#include <linux/module.h> 21#include <linux/percpu.h> 22#include <linux/mutex.h> 23#include <linux/delay.h> 24#include <linux/slab.h> 25#include <linux/init.h> 26#include <linux/hash.h> 27#include <linux/list.h> 28#include <linux/cpu.h> 29#include <linux/oom.h> 30#include <linux/mm.h> 31 32#include <asm/local64.h> 33#include <asm/local.h> 34#include <asm/setup.h> 35 36#include "trace.h" 37 38/* 39 * The "absolute" timestamp in the buffer is only 59 bits. 40 * If a clock has the 5 MSBs set, it needs to be saved and 41 * reinserted. 42 */ 43#define TS_MSB (0xf8ULL << 56) 44#define ABS_TS_MASK (~TS_MSB) 45 46static void update_pages_handler(struct work_struct *work); 47 48#define RING_BUFFER_META_MAGIC 0xBADFEED 49 50struct ring_buffer_meta { 51 int magic; 52 int struct_sizes; 53 unsigned long total_size; 54 unsigned long buffers_offset; 55}; 56 57struct ring_buffer_cpu_meta { 58 unsigned long first_buffer; 59 unsigned long head_buffer; 60 unsigned long commit_buffer; 61 __u32 subbuf_size; 62 __u32 nr_subbufs; 63 int buffers[]; 64}; 65 66/* 67 * The ring buffer header is special. We must manually up keep it. 68 */ 69int ring_buffer_print_entry_header(struct trace_seq *s) 70{ 71 trace_seq_puts(s, "# compressed entry header\n"); 72 trace_seq_puts(s, "\ttype_len : 5 bits\n"); 73 trace_seq_puts(s, "\ttime_delta : 27 bits\n"); 74 trace_seq_puts(s, "\tarray : 32 bits\n"); 75 trace_seq_putc(s, '\n'); 76 trace_seq_printf(s, "\tpadding : type == %d\n", 77 RINGBUF_TYPE_PADDING); 78 trace_seq_printf(s, "\ttime_extend : type == %d\n", 79 RINGBUF_TYPE_TIME_EXTEND); 80 trace_seq_printf(s, "\ttime_stamp : type == %d\n", 81 RINGBUF_TYPE_TIME_STAMP); 82 trace_seq_printf(s, "\tdata max type_len == %d\n", 83 RINGBUF_TYPE_DATA_TYPE_LEN_MAX); 84 85 return !trace_seq_has_overflowed(s); 86} 87 88/* 89 * The ring buffer is made up of a list of pages. A separate list of pages is 90 * allocated for each CPU. A writer may only write to a buffer that is 91 * associated with the CPU it is currently executing on. A reader may read 92 * from any per cpu buffer. 93 * 94 * The reader is special. For each per cpu buffer, the reader has its own 95 * reader page. When a reader has read the entire reader page, this reader 96 * page is swapped with another page in the ring buffer. 97 * 98 * Now, as long as the writer is off the reader page, the reader can do what 99 * ever it wants with that page. The writer will never write to that page 100 * again (as long as it is out of the ring buffer). 101 * 102 * Here's some silly ASCII art. 103 * 104 * +------+ 105 * |reader| RING BUFFER 106 * |page | 107 * +------+ +---+ +---+ +---+ 108 * | |-->| |-->| | 109 * +---+ +---+ +---+ 110 * ^ | 111 * | | 112 * +---------------+ 113 * 114 * 115 * +------+ 116 * |reader| RING BUFFER 117 * |page |------------------v 118 * +------+ +---+ +---+ +---+ 119 * | |-->| |-->| | 120 * +---+ +---+ +---+ 121 * ^ | 122 * | | 123 * +---------------+ 124 * 125 * 126 * +------+ 127 * |reader| RING BUFFER 128 * |page |------------------v 129 * +------+ +---+ +---+ +---+ 130 * ^ | |-->| |-->| | 131 * | +---+ +---+ +---+ 132 * | | 133 * | | 134 * +------------------------------+ 135 * 136 * 137 * +------+ 138 * |buffer| RING BUFFER 139 * |page |------------------v 140 * +------+ +---+ +---+ +---+ 141 * ^ | | | |-->| | 142 * | New +---+ +---+ +---+ 143 * | Reader------^ | 144 * | page | 145 * +------------------------------+ 146 * 147 * 148 * After we make this swap, the reader can hand this page off to the splice 149 * code and be done with it. It can even allocate a new page if it needs to 150 * and swap that into the ring buffer. 151 * 152 * We will be using cmpxchg soon to make all this lockless. 153 * 154 */ 155 156/* Used for individual buffers (after the counter) */ 157#define RB_BUFFER_OFF (1 << 20) 158 159#define BUF_PAGE_HDR_SIZE offsetof(struct buffer_data_page, data) 160 161#define RB_EVNT_HDR_SIZE (offsetof(struct ring_buffer_event, array)) 162#define RB_ALIGNMENT 4U 163#define RB_MAX_SMALL_DATA (RB_ALIGNMENT * RINGBUF_TYPE_DATA_TYPE_LEN_MAX) 164#define RB_EVNT_MIN_SIZE 8U /* two 32bit words */ 165 166#ifndef CONFIG_HAVE_64BIT_ALIGNED_ACCESS 167# define RB_FORCE_8BYTE_ALIGNMENT 0 168# define RB_ARCH_ALIGNMENT RB_ALIGNMENT 169#else 170# define RB_FORCE_8BYTE_ALIGNMENT 1 171# define RB_ARCH_ALIGNMENT 8U 172#endif 173 174#define RB_ALIGN_DATA __aligned(RB_ARCH_ALIGNMENT) 175 176/* define RINGBUF_TYPE_DATA for 'case RINGBUF_TYPE_DATA:' */ 177#define RINGBUF_TYPE_DATA 0 ... RINGBUF_TYPE_DATA_TYPE_LEN_MAX 178 179enum { 180 RB_LEN_TIME_EXTEND = 8, 181 RB_LEN_TIME_STAMP = 8, 182}; 183 184#define skip_time_extend(event) \ 185 ((struct ring_buffer_event *)((char *)event + RB_LEN_TIME_EXTEND)) 186 187#define extended_time(event) \ 188 (event->type_len >= RINGBUF_TYPE_TIME_EXTEND) 189 190static inline bool rb_null_event(struct ring_buffer_event *event) 191{ 192 return event->type_len == RINGBUF_TYPE_PADDING && !event->time_delta; 193} 194 195static void rb_event_set_padding(struct ring_buffer_event *event) 196{ 197 /* padding has a NULL time_delta */ 198 event->type_len = RINGBUF_TYPE_PADDING; 199 event->time_delta = 0; 200} 201 202static unsigned 203rb_event_data_length(struct ring_buffer_event *event) 204{ 205 unsigned length; 206 207 if (event->type_len) 208 length = event->type_len * RB_ALIGNMENT; 209 else 210 length = event->array[0]; 211 return length + RB_EVNT_HDR_SIZE; 212} 213 214/* 215 * Return the length of the given event. Will return 216 * the length of the time extend if the event is a 217 * time extend. 218 */ 219static inline unsigned 220rb_event_length(struct ring_buffer_event *event) 221{ 222 switch (event->type_len) { 223 case RINGBUF_TYPE_PADDING: 224 if (rb_null_event(event)) 225 /* undefined */ 226 return -1; 227 return event->array[0] + RB_EVNT_HDR_SIZE; 228 229 case RINGBUF_TYPE_TIME_EXTEND: 230 return RB_LEN_TIME_EXTEND; 231 232 case RINGBUF_TYPE_TIME_STAMP: 233 return RB_LEN_TIME_STAMP; 234 235 case RINGBUF_TYPE_DATA: 236 return rb_event_data_length(event); 237 default: 238 WARN_ON_ONCE(1); 239 } 240 /* not hit */ 241 return 0; 242} 243 244/* 245 * Return total length of time extend and data, 246 * or just the event length for all other events. 247 */ 248static inline unsigned 249rb_event_ts_length(struct ring_buffer_event *event) 250{ 251 unsigned len = 0; 252 253 if (extended_time(event)) { 254 /* time extends include the data event after it */ 255 len = RB_LEN_TIME_EXTEND; 256 event = skip_time_extend(event); 257 } 258 return len + rb_event_length(event); 259} 260 261/** 262 * ring_buffer_event_length - return the length of the event 263 * @event: the event to get the length of 264 * 265 * Returns the size of the data load of a data event. 266 * If the event is something other than a data event, it 267 * returns the size of the event itself. With the exception 268 * of a TIME EXTEND, where it still returns the size of the 269 * data load of the data event after it. 270 */ 271unsigned ring_buffer_event_length(struct ring_buffer_event *event) 272{ 273 unsigned length; 274 275 if (extended_time(event)) 276 event = skip_time_extend(event); 277 278 length = rb_event_length(event); 279 if (event->type_len > RINGBUF_TYPE_DATA_TYPE_LEN_MAX) 280 return length; 281 length -= RB_EVNT_HDR_SIZE; 282 if (length > RB_MAX_SMALL_DATA + sizeof(event->array[0])) 283 length -= sizeof(event->array[0]); 284 return length; 285} 286EXPORT_SYMBOL_GPL(ring_buffer_event_length); 287 288/* inline for ring buffer fast paths */ 289static __always_inline void * 290rb_event_data(struct ring_buffer_event *event) 291{ 292 if (extended_time(event)) 293 event = skip_time_extend(event); 294 WARN_ON_ONCE(event->type_len > RINGBUF_TYPE_DATA_TYPE_LEN_MAX); 295 /* If length is in len field, then array[0] has the data */ 296 if (event->type_len) 297 return (void *)&event->array[0]; 298 /* Otherwise length is in array[0] and array[1] has the data */ 299 return (void *)&event->array[1]; 300} 301 302/** 303 * ring_buffer_event_data - return the data of the event 304 * @event: the event to get the data from 305 */ 306void *ring_buffer_event_data(struct ring_buffer_event *event) 307{ 308 return rb_event_data(event); 309} 310EXPORT_SYMBOL_GPL(ring_buffer_event_data); 311 312#define for_each_buffer_cpu(buffer, cpu) \ 313 for_each_cpu(cpu, buffer->cpumask) 314 315#define for_each_online_buffer_cpu(buffer, cpu) \ 316 for_each_cpu_and(cpu, buffer->cpumask, cpu_online_mask) 317 318#define TS_SHIFT 27 319#define TS_MASK ((1ULL << TS_SHIFT) - 1) 320#define TS_DELTA_TEST (~TS_MASK) 321 322static u64 rb_event_time_stamp(struct ring_buffer_event *event) 323{ 324 u64 ts; 325 326 ts = event->array[0]; 327 ts <<= TS_SHIFT; 328 ts += event->time_delta; 329 330 return ts; 331} 332 333/* Flag when events were overwritten */ 334#define RB_MISSED_EVENTS (1 << 31) 335/* Missed count stored at end */ 336#define RB_MISSED_STORED (1 << 30) 337 338#define RB_MISSED_MASK (3 << 30) 339 340struct buffer_data_page { 341 u64 time_stamp; /* page time stamp */ 342 local_t commit; /* write committed index */ 343 unsigned char data[] RB_ALIGN_DATA; /* data of buffer page */ 344}; 345 346struct buffer_data_read_page { 347 unsigned order; /* order of the page */ 348 struct buffer_data_page *data; /* actual data, stored in this page */ 349}; 350 351/* 352 * Note, the buffer_page list must be first. The buffer pages 353 * are allocated in cache lines, which means that each buffer 354 * page will be at the beginning of a cache line, and thus 355 * the least significant bits will be zero. We use this to 356 * add flags in the list struct pointers, to make the ring buffer 357 * lockless. 358 */ 359struct buffer_page { 360 struct list_head list; /* list of buffer pages */ 361 local_t write; /* index for next write */ 362 unsigned read; /* index for next read */ 363 local_t entries; /* entries on this page */ 364 unsigned long real_end; /* real end of data */ 365 unsigned order; /* order of the page */ 366 u32 id:30; /* ID for external mapping */ 367 u32 range:1; /* Mapped via a range */ 368 struct buffer_data_page *page; /* Actual data page */ 369}; 370 371/* 372 * The buffer page counters, write and entries, must be reset 373 * atomically when crossing page boundaries. To synchronize this 374 * update, two counters are inserted into the number. One is 375 * the actual counter for the write position or count on the page. 376 * 377 * The other is a counter of updaters. Before an update happens 378 * the update partition of the counter is incremented. This will 379 * allow the updater to update the counter atomically. 380 * 381 * The counter is 20 bits, and the state data is 12. 382 */ 383#define RB_WRITE_MASK 0xfffff 384#define RB_WRITE_INTCNT (1 << 20) 385 386static void rb_init_page(struct buffer_data_page *bpage) 387{ 388 local_set(&bpage->commit, 0); 389} 390 391static __always_inline unsigned int rb_page_commit(struct buffer_page *bpage) 392{ 393 return local_read(&bpage->page->commit); 394} 395 396static void free_buffer_page(struct buffer_page *bpage) 397{ 398 /* Range pages are not to be freed */ 399 if (!bpage->range) 400 free_pages((unsigned long)bpage->page, bpage->order); 401 kfree(bpage); 402} 403 404/* 405 * For best performance, allocate cpu buffer data cache line sized 406 * and per CPU. 407 */ 408#define alloc_cpu_buffer(cpu) (struct ring_buffer_per_cpu *) \ 409 kzalloc_node(ALIGN(sizeof(struct ring_buffer_per_cpu), \ 410 cache_line_size()), GFP_KERNEL, cpu_to_node(cpu)); 411 412#define alloc_cpu_page(cpu) (struct buffer_page *) \ 413 kzalloc_node(ALIGN(sizeof(struct buffer_page), \ 414 cache_line_size()), GFP_KERNEL, cpu_to_node(cpu)); 415 416static struct buffer_data_page *alloc_cpu_data(int cpu, int order) 417{ 418 struct buffer_data_page *dpage; 419 struct page *page; 420 gfp_t mflags; 421 422 /* 423 * __GFP_RETRY_MAYFAIL flag makes sure that the allocation fails 424 * gracefully without invoking oom-killer and the system is not 425 * destabilized. 426 */ 427 mflags = GFP_KERNEL | __GFP_RETRY_MAYFAIL | __GFP_COMP | __GFP_ZERO; 428 429 page = alloc_pages_node(cpu_to_node(cpu), mflags, order); 430 if (!page) 431 return NULL; 432 433 dpage = page_address(page); 434 rb_init_page(dpage); 435 436 return dpage; 437} 438 439/* 440 * We need to fit the time_stamp delta into 27 bits. 441 */ 442static inline bool test_time_stamp(u64 delta) 443{ 444 return !!(delta & TS_DELTA_TEST); 445} 446 447struct rb_irq_work { 448 struct irq_work work; 449 wait_queue_head_t waiters; 450 wait_queue_head_t full_waiters; 451 atomic_t seq; 452 bool waiters_pending; 453 bool full_waiters_pending; 454 bool wakeup_full; 455}; 456 457/* 458 * Structure to hold event state and handle nested events. 459 */ 460struct rb_event_info { 461 u64 ts; 462 u64 delta; 463 u64 before; 464 u64 after; 465 unsigned long length; 466 struct buffer_page *tail_page; 467 int add_timestamp; 468}; 469 470/* 471 * Used for the add_timestamp 472 * NONE 473 * EXTEND - wants a time extend 474 * ABSOLUTE - the buffer requests all events to have absolute time stamps 475 * FORCE - force a full time stamp. 476 */ 477enum { 478 RB_ADD_STAMP_NONE = 0, 479 RB_ADD_STAMP_EXTEND = BIT(1), 480 RB_ADD_STAMP_ABSOLUTE = BIT(2), 481 RB_ADD_STAMP_FORCE = BIT(3) 482}; 483/* 484 * Used for which event context the event is in. 485 * TRANSITION = 0 486 * NMI = 1 487 * IRQ = 2 488 * SOFTIRQ = 3 489 * NORMAL = 4 490 * 491 * See trace_recursive_lock() comment below for more details. 492 */ 493enum { 494 RB_CTX_TRANSITION, 495 RB_CTX_NMI, 496 RB_CTX_IRQ, 497 RB_CTX_SOFTIRQ, 498 RB_CTX_NORMAL, 499 RB_CTX_MAX 500}; 501 502struct rb_time_struct { 503 local64_t time; 504}; 505typedef struct rb_time_struct rb_time_t; 506 507#define MAX_NEST 5 508 509/* 510 * head_page == tail_page && head == tail then buffer is empty. 511 */ 512struct ring_buffer_per_cpu { 513 int cpu; 514 atomic_t record_disabled; 515 atomic_t resize_disabled; 516 struct trace_buffer *buffer; 517 raw_spinlock_t reader_lock; /* serialize readers */ 518 arch_spinlock_t lock; 519 struct lock_class_key lock_key; 520 struct buffer_data_page *free_page; 521 unsigned long nr_pages; 522 unsigned int current_context; 523 struct list_head *pages; 524 /* pages generation counter, incremented when the list changes */ 525 unsigned long cnt; 526 struct buffer_page *head_page; /* read from head */ 527 struct buffer_page *tail_page; /* write to tail */ 528 struct buffer_page *commit_page; /* committed pages */ 529 struct buffer_page *reader_page; 530 unsigned long lost_events; 531 unsigned long last_overrun; 532 unsigned long nest; 533 local_t entries_bytes; 534 local_t entries; 535 local_t overrun; 536 local_t commit_overrun; 537 local_t dropped_events; 538 local_t committing; 539 local_t commits; 540 local_t pages_touched; 541 local_t pages_lost; 542 local_t pages_read; 543 long last_pages_touch; 544 size_t shortest_full; 545 unsigned long read; 546 unsigned long read_bytes; 547 rb_time_t write_stamp; 548 rb_time_t before_stamp; 549 u64 event_stamp[MAX_NEST]; 550 u64 read_stamp; 551 /* pages removed since last reset */ 552 unsigned long pages_removed; 553 554 unsigned int mapped; 555 unsigned int user_mapped; /* user space mapping */ 556 struct mutex mapping_lock; 557 unsigned long *subbuf_ids; /* ID to subbuf VA */ 558 struct trace_buffer_meta *meta_page; 559 struct ring_buffer_cpu_meta *ring_meta; 560 561 /* ring buffer pages to update, > 0 to add, < 0 to remove */ 562 long nr_pages_to_update; 563 struct list_head new_pages; /* new pages to add */ 564 struct work_struct update_pages_work; 565 struct completion update_done; 566 567 struct rb_irq_work irq_work; 568}; 569 570struct trace_buffer { 571 unsigned flags; 572 int cpus; 573 atomic_t record_disabled; 574 atomic_t resizing; 575 cpumask_var_t cpumask; 576 577 struct lock_class_key *reader_lock_key; 578 579 struct mutex mutex; 580 581 struct ring_buffer_per_cpu **buffers; 582 583 struct hlist_node node; 584 u64 (*clock)(void); 585 586 struct rb_irq_work irq_work; 587 bool time_stamp_abs; 588 589 unsigned long range_addr_start; 590 unsigned long range_addr_end; 591 592 struct ring_buffer_meta *meta; 593 594 unsigned int subbuf_size; 595 unsigned int subbuf_order; 596 unsigned int max_data_size; 597}; 598 599struct ring_buffer_iter { 600 struct ring_buffer_per_cpu *cpu_buffer; 601 unsigned long head; 602 unsigned long next_event; 603 struct buffer_page *head_page; 604 struct buffer_page *cache_reader_page; 605 unsigned long cache_read; 606 unsigned long cache_pages_removed; 607 u64 read_stamp; 608 u64 page_stamp; 609 struct ring_buffer_event *event; 610 size_t event_size; 611 int missed_events; 612}; 613 614int ring_buffer_print_page_header(struct trace_buffer *buffer, struct trace_seq *s) 615{ 616 struct buffer_data_page field; 617 618 trace_seq_printf(s, "\tfield: u64 timestamp;\t" 619 "offset:0;\tsize:%u;\tsigned:%u;\n", 620 (unsigned int)sizeof(field.time_stamp), 621 (unsigned int)is_signed_type(u64)); 622 623 trace_seq_printf(s, "\tfield: local_t commit;\t" 624 "offset:%u;\tsize:%u;\tsigned:%u;\n", 625 (unsigned int)offsetof(typeof(field), commit), 626 (unsigned int)sizeof(field.commit), 627 (unsigned int)is_signed_type(long)); 628 629 trace_seq_printf(s, "\tfield: int overwrite;\t" 630 "offset:%u;\tsize:%u;\tsigned:%u;\n", 631 (unsigned int)offsetof(typeof(field), commit), 632 1, 633 (unsigned int)is_signed_type(long)); 634 635 trace_seq_printf(s, "\tfield: char data;\t" 636 "offset:%u;\tsize:%u;\tsigned:%u;\n", 637 (unsigned int)offsetof(typeof(field), data), 638 (unsigned int)buffer->subbuf_size, 639 (unsigned int)is_signed_type(char)); 640 641 return !trace_seq_has_overflowed(s); 642} 643 644static inline void rb_time_read(rb_time_t *t, u64 *ret) 645{ 646 *ret = local64_read(&t->time); 647} 648static void rb_time_set(rb_time_t *t, u64 val) 649{ 650 local64_set(&t->time, val); 651} 652 653/* 654 * Enable this to make sure that the event passed to 655 * ring_buffer_event_time_stamp() is not committed and also 656 * is on the buffer that it passed in. 657 */ 658//#define RB_VERIFY_EVENT 659#ifdef RB_VERIFY_EVENT 660static struct list_head *rb_list_head(struct list_head *list); 661static void verify_event(struct ring_buffer_per_cpu *cpu_buffer, 662 void *event) 663{ 664 struct buffer_page *page = cpu_buffer->commit_page; 665 struct buffer_page *tail_page = READ_ONCE(cpu_buffer->tail_page); 666 struct list_head *next; 667 long commit, write; 668 unsigned long addr = (unsigned long)event; 669 bool done = false; 670 int stop = 0; 671 672 /* Make sure the event exists and is not committed yet */ 673 do { 674 if (page == tail_page || WARN_ON_ONCE(stop++ > 100)) 675 done = true; 676 commit = local_read(&page->page->commit); 677 write = local_read(&page->write); 678 if (addr >= (unsigned long)&page->page->data[commit] && 679 addr < (unsigned long)&page->page->data[write]) 680 return; 681 682 next = rb_list_head(page->list.next); 683 page = list_entry(next, struct buffer_page, list); 684 } while (!done); 685 WARN_ON_ONCE(1); 686} 687#else 688static inline void verify_event(struct ring_buffer_per_cpu *cpu_buffer, 689 void *event) 690{ 691} 692#endif 693 694/* 695 * The absolute time stamp drops the 5 MSBs and some clocks may 696 * require them. The rb_fix_abs_ts() will take a previous full 697 * time stamp, and add the 5 MSB of that time stamp on to the 698 * saved absolute time stamp. Then they are compared in case of 699 * the unlikely event that the latest time stamp incremented 700 * the 5 MSB. 701 */ 702static inline u64 rb_fix_abs_ts(u64 abs, u64 save_ts) 703{ 704 if (save_ts & TS_MSB) { 705 abs |= save_ts & TS_MSB; 706 /* Check for overflow */ 707 if (unlikely(abs < save_ts)) 708 abs += 1ULL << 59; 709 } 710 return abs; 711} 712 713static inline u64 rb_time_stamp(struct trace_buffer *buffer); 714 715/** 716 * ring_buffer_event_time_stamp - return the event's current time stamp 717 * @buffer: The buffer that the event is on 718 * @event: the event to get the time stamp of 719 * 720 * Note, this must be called after @event is reserved, and before it is 721 * committed to the ring buffer. And must be called from the same 722 * context where the event was reserved (normal, softirq, irq, etc). 723 * 724 * Returns the time stamp associated with the current event. 725 * If the event has an extended time stamp, then that is used as 726 * the time stamp to return. 727 * In the highly unlikely case that the event was nested more than 728 * the max nesting, then the write_stamp of the buffer is returned, 729 * otherwise current time is returned, but that really neither of 730 * the last two cases should ever happen. 731 */ 732u64 ring_buffer_event_time_stamp(struct trace_buffer *buffer, 733 struct ring_buffer_event *event) 734{ 735 struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[smp_processor_id()]; 736 unsigned int nest; 737 u64 ts; 738 739 /* If the event includes an absolute time, then just use that */ 740 if (event->type_len == RINGBUF_TYPE_TIME_STAMP) { 741 ts = rb_event_time_stamp(event); 742 return rb_fix_abs_ts(ts, cpu_buffer->tail_page->page->time_stamp); 743 } 744 745 nest = local_read(&cpu_buffer->committing); 746 verify_event(cpu_buffer, event); 747 if (WARN_ON_ONCE(!nest)) 748 goto fail; 749 750 /* Read the current saved nesting level time stamp */ 751 if (likely(--nest < MAX_NEST)) 752 return cpu_buffer->event_stamp[nest]; 753 754 /* Shouldn't happen, warn if it does */ 755 WARN_ONCE(1, "nest (%d) greater than max", nest); 756 757 fail: 758 rb_time_read(&cpu_buffer->write_stamp, &ts); 759 760 return ts; 761} 762 763/** 764 * ring_buffer_nr_dirty_pages - get the number of used pages in the ring buffer 765 * @buffer: The ring_buffer to get the number of pages from 766 * @cpu: The cpu of the ring_buffer to get the number of pages from 767 * 768 * Returns the number of pages that have content in the ring buffer. 769 */ 770size_t ring_buffer_nr_dirty_pages(struct trace_buffer *buffer, int cpu) 771{ 772 size_t read; 773 size_t lost; 774 size_t cnt; 775 776 read = local_read(&buffer->buffers[cpu]->pages_read); 777 lost = local_read(&buffer->buffers[cpu]->pages_lost); 778 cnt = local_read(&buffer->buffers[cpu]->pages_touched); 779 780 if (WARN_ON_ONCE(cnt < lost)) 781 return 0; 782 783 cnt -= lost; 784 785 /* The reader can read an empty page, but not more than that */ 786 if (cnt < read) { 787 WARN_ON_ONCE(read > cnt + 1); 788 return 0; 789 } 790 791 return cnt - read; 792} 793 794static __always_inline bool full_hit(struct trace_buffer *buffer, int cpu, int full) 795{ 796 struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu]; 797 size_t nr_pages; 798 size_t dirty; 799 800 nr_pages = cpu_buffer->nr_pages; 801 if (!nr_pages || !full) 802 return true; 803 804 /* 805 * Add one as dirty will never equal nr_pages, as the sub-buffer 806 * that the writer is on is not counted as dirty. 807 * This is needed if "buffer_percent" is set to 100. 808 */ 809 dirty = ring_buffer_nr_dirty_pages(buffer, cpu) + 1; 810 811 return (dirty * 100) >= (full * nr_pages); 812} 813 814/* 815 * rb_wake_up_waiters - wake up tasks waiting for ring buffer input 816 * 817 * Schedules a delayed work to wake up any task that is blocked on the 818 * ring buffer waiters queue. 819 */ 820static void rb_wake_up_waiters(struct irq_work *work) 821{ 822 struct rb_irq_work *rbwork = container_of(work, struct rb_irq_work, work); 823 824 /* For waiters waiting for the first wake up */ 825 (void)atomic_fetch_inc_release(&rbwork->seq); 826 827 wake_up_all(&rbwork->waiters); 828 if (rbwork->full_waiters_pending || rbwork->wakeup_full) { 829 /* Only cpu_buffer sets the above flags */ 830 struct ring_buffer_per_cpu *cpu_buffer = 831 container_of(rbwork, struct ring_buffer_per_cpu, irq_work); 832 833 /* Called from interrupt context */ 834 raw_spin_lock(&cpu_buffer->reader_lock); 835 rbwork->wakeup_full = false; 836 rbwork->full_waiters_pending = false; 837 838 /* Waking up all waiters, they will reset the shortest full */ 839 cpu_buffer->shortest_full = 0; 840 raw_spin_unlock(&cpu_buffer->reader_lock); 841 842 wake_up_all(&rbwork->full_waiters); 843 } 844} 845 846/** 847 * ring_buffer_wake_waiters - wake up any waiters on this ring buffer 848 * @buffer: The ring buffer to wake waiters on 849 * @cpu: The CPU buffer to wake waiters on 850 * 851 * In the case of a file that represents a ring buffer is closing, 852 * it is prudent to wake up any waiters that are on this. 853 */ 854void ring_buffer_wake_waiters(struct trace_buffer *buffer, int cpu) 855{ 856 struct ring_buffer_per_cpu *cpu_buffer; 857 struct rb_irq_work *rbwork; 858 859 if (!buffer) 860 return; 861 862 if (cpu == RING_BUFFER_ALL_CPUS) { 863 864 /* Wake up individual ones too. One level recursion */ 865 for_each_buffer_cpu(buffer, cpu) 866 ring_buffer_wake_waiters(buffer, cpu); 867 868 rbwork = &buffer->irq_work; 869 } else { 870 if (WARN_ON_ONCE(!buffer->buffers)) 871 return; 872 if (WARN_ON_ONCE(cpu >= nr_cpu_ids)) 873 return; 874 875 cpu_buffer = buffer->buffers[cpu]; 876 /* The CPU buffer may not have been initialized yet */ 877 if (!cpu_buffer) 878 return; 879 rbwork = &cpu_buffer->irq_work; 880 } 881 882 /* This can be called in any context */ 883 irq_work_queue(&rbwork->work); 884} 885 886static bool rb_watermark_hit(struct trace_buffer *buffer, int cpu, int full) 887{ 888 struct ring_buffer_per_cpu *cpu_buffer; 889 bool ret = false; 890 891 /* Reads of all CPUs always waits for any data */ 892 if (cpu == RING_BUFFER_ALL_CPUS) 893 return !ring_buffer_empty(buffer); 894 895 cpu_buffer = buffer->buffers[cpu]; 896 897 if (!ring_buffer_empty_cpu(buffer, cpu)) { 898 unsigned long flags; 899 bool pagebusy; 900 901 if (!full) 902 return true; 903 904 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 905 pagebusy = cpu_buffer->reader_page == cpu_buffer->commit_page; 906 ret = !pagebusy && full_hit(buffer, cpu, full); 907 908 if (!ret && (!cpu_buffer->shortest_full || 909 cpu_buffer->shortest_full > full)) { 910 cpu_buffer->shortest_full = full; 911 } 912 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 913 } 914 return ret; 915} 916 917static inline bool 918rb_wait_cond(struct rb_irq_work *rbwork, struct trace_buffer *buffer, 919 int cpu, int full, ring_buffer_cond_fn cond, void *data) 920{ 921 if (rb_watermark_hit(buffer, cpu, full)) 922 return true; 923 924 if (cond(data)) 925 return true; 926 927 /* 928 * The events can happen in critical sections where 929 * checking a work queue can cause deadlocks. 930 * After adding a task to the queue, this flag is set 931 * only to notify events to try to wake up the queue 932 * using irq_work. 933 * 934 * We don't clear it even if the buffer is no longer 935 * empty. The flag only causes the next event to run 936 * irq_work to do the work queue wake up. The worse 937 * that can happen if we race with !trace_empty() is that 938 * an event will cause an irq_work to try to wake up 939 * an empty queue. 940 * 941 * There's no reason to protect this flag either, as 942 * the work queue and irq_work logic will do the necessary 943 * synchronization for the wake ups. The only thing 944 * that is necessary is that the wake up happens after 945 * a task has been queued. It's OK for spurious wake ups. 946 */ 947 if (full) 948 rbwork->full_waiters_pending = true; 949 else 950 rbwork->waiters_pending = true; 951 952 return false; 953} 954 955struct rb_wait_data { 956 struct rb_irq_work *irq_work; 957 int seq; 958}; 959 960/* 961 * The default wait condition for ring_buffer_wait() is to just to exit the 962 * wait loop the first time it is woken up. 963 */ 964static bool rb_wait_once(void *data) 965{ 966 struct rb_wait_data *rdata = data; 967 struct rb_irq_work *rbwork = rdata->irq_work; 968 969 return atomic_read_acquire(&rbwork->seq) != rdata->seq; 970} 971 972/** 973 * ring_buffer_wait - wait for input to the ring buffer 974 * @buffer: buffer to wait on 975 * @cpu: the cpu buffer to wait on 976 * @full: wait until the percentage of pages are available, if @cpu != RING_BUFFER_ALL_CPUS 977 * @cond: condition function to break out of wait (NULL to run once) 978 * @data: the data to pass to @cond. 979 * 980 * If @cpu == RING_BUFFER_ALL_CPUS then the task will wake up as soon 981 * as data is added to any of the @buffer's cpu buffers. Otherwise 982 * it will wait for data to be added to a specific cpu buffer. 983 */ 984int ring_buffer_wait(struct trace_buffer *buffer, int cpu, int full, 985 ring_buffer_cond_fn cond, void *data) 986{ 987 struct ring_buffer_per_cpu *cpu_buffer; 988 struct wait_queue_head *waitq; 989 struct rb_irq_work *rbwork; 990 struct rb_wait_data rdata; 991 int ret = 0; 992 993 /* 994 * Depending on what the caller is waiting for, either any 995 * data in any cpu buffer, or a specific buffer, put the 996 * caller on the appropriate wait queue. 997 */ 998 if (cpu == RING_BUFFER_ALL_CPUS) { 999 rbwork = &buffer->irq_work; 1000 /* Full only makes sense on per cpu reads */ 1001 full = 0; 1002 } else { 1003 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 1004 return -ENODEV; 1005 cpu_buffer = buffer->buffers[cpu]; 1006 rbwork = &cpu_buffer->irq_work; 1007 } 1008 1009 if (full) 1010 waitq = &rbwork->full_waiters; 1011 else 1012 waitq = &rbwork->waiters; 1013 1014 /* Set up to exit loop as soon as it is woken */ 1015 if (!cond) { 1016 cond = rb_wait_once; 1017 rdata.irq_work = rbwork; 1018 rdata.seq = atomic_read_acquire(&rbwork->seq); 1019 data = &rdata; 1020 } 1021 1022 ret = wait_event_interruptible((*waitq), 1023 rb_wait_cond(rbwork, buffer, cpu, full, cond, data)); 1024 1025 return ret; 1026} 1027 1028/** 1029 * ring_buffer_poll_wait - poll on buffer input 1030 * @buffer: buffer to wait on 1031 * @cpu: the cpu buffer to wait on 1032 * @filp: the file descriptor 1033 * @poll_table: The poll descriptor 1034 * @full: wait until the percentage of pages are available, if @cpu != RING_BUFFER_ALL_CPUS 1035 * 1036 * If @cpu == RING_BUFFER_ALL_CPUS then the task will wake up as soon 1037 * as data is added to any of the @buffer's cpu buffers. Otherwise 1038 * it will wait for data to be added to a specific cpu buffer. 1039 * 1040 * Returns EPOLLIN | EPOLLRDNORM if data exists in the buffers, 1041 * zero otherwise. 1042 */ 1043__poll_t ring_buffer_poll_wait(struct trace_buffer *buffer, int cpu, 1044 struct file *filp, poll_table *poll_table, int full) 1045{ 1046 struct ring_buffer_per_cpu *cpu_buffer; 1047 struct rb_irq_work *rbwork; 1048 1049 if (cpu == RING_BUFFER_ALL_CPUS) { 1050 rbwork = &buffer->irq_work; 1051 full = 0; 1052 } else { 1053 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 1054 return EPOLLERR; 1055 1056 cpu_buffer = buffer->buffers[cpu]; 1057 rbwork = &cpu_buffer->irq_work; 1058 } 1059 1060 if (full) { 1061 poll_wait(filp, &rbwork->full_waiters, poll_table); 1062 1063 if (rb_watermark_hit(buffer, cpu, full)) 1064 return EPOLLIN | EPOLLRDNORM; 1065 /* 1066 * Only allow full_waiters_pending update to be seen after 1067 * the shortest_full is set (in rb_watermark_hit). If the 1068 * writer sees the full_waiters_pending flag set, it will 1069 * compare the amount in the ring buffer to shortest_full. 1070 * If the amount in the ring buffer is greater than the 1071 * shortest_full percent, it will call the irq_work handler 1072 * to wake up this list. The irq_handler will reset shortest_full 1073 * back to zero. That's done under the reader_lock, but 1074 * the below smp_mb() makes sure that the update to 1075 * full_waiters_pending doesn't leak up into the above. 1076 */ 1077 smp_mb(); 1078 rbwork->full_waiters_pending = true; 1079 return 0; 1080 } 1081 1082 poll_wait(filp, &rbwork->waiters, poll_table); 1083 rbwork->waiters_pending = true; 1084 1085 /* 1086 * There's a tight race between setting the waiters_pending and 1087 * checking if the ring buffer is empty. Once the waiters_pending bit 1088 * is set, the next event will wake the task up, but we can get stuck 1089 * if there's only a single event in. 1090 * 1091 * FIXME: Ideally, we need a memory barrier on the writer side as well, 1092 * but adding a memory barrier to all events will cause too much of a 1093 * performance hit in the fast path. We only need a memory barrier when 1094 * the buffer goes from empty to having content. But as this race is 1095 * extremely small, and it's not a problem if another event comes in, we 1096 * will fix it later. 1097 */ 1098 smp_mb(); 1099 1100 if ((cpu == RING_BUFFER_ALL_CPUS && !ring_buffer_empty(buffer)) || 1101 (cpu != RING_BUFFER_ALL_CPUS && !ring_buffer_empty_cpu(buffer, cpu))) 1102 return EPOLLIN | EPOLLRDNORM; 1103 return 0; 1104} 1105 1106/* buffer may be either ring_buffer or ring_buffer_per_cpu */ 1107#define RB_WARN_ON(b, cond) \ 1108 ({ \ 1109 int _____ret = unlikely(cond); \ 1110 if (_____ret) { \ 1111 if (__same_type(*(b), struct ring_buffer_per_cpu)) { \ 1112 struct ring_buffer_per_cpu *__b = \ 1113 (void *)b; \ 1114 atomic_inc(&__b->buffer->record_disabled); \ 1115 } else \ 1116 atomic_inc(&b->record_disabled); \ 1117 WARN_ON(1); \ 1118 } \ 1119 _____ret; \ 1120 }) 1121 1122/* Up this if you want to test the TIME_EXTENTS and normalization */ 1123#define DEBUG_SHIFT 0 1124 1125static inline u64 rb_time_stamp(struct trace_buffer *buffer) 1126{ 1127 u64 ts; 1128 1129 /* Skip retpolines :-( */ 1130 if (IS_ENABLED(CONFIG_MITIGATION_RETPOLINE) && likely(buffer->clock == trace_clock_local)) 1131 ts = trace_clock_local(); 1132 else 1133 ts = buffer->clock(); 1134 1135 /* shift to debug/test normalization and TIME_EXTENTS */ 1136 return ts << DEBUG_SHIFT; 1137} 1138 1139u64 ring_buffer_time_stamp(struct trace_buffer *buffer) 1140{ 1141 u64 time; 1142 1143 preempt_disable_notrace(); 1144 time = rb_time_stamp(buffer); 1145 preempt_enable_notrace(); 1146 1147 return time; 1148} 1149EXPORT_SYMBOL_GPL(ring_buffer_time_stamp); 1150 1151void ring_buffer_normalize_time_stamp(struct trace_buffer *buffer, 1152 int cpu, u64 *ts) 1153{ 1154 /* Just stupid testing the normalize function and deltas */ 1155 *ts >>= DEBUG_SHIFT; 1156} 1157EXPORT_SYMBOL_GPL(ring_buffer_normalize_time_stamp); 1158 1159/* 1160 * Making the ring buffer lockless makes things tricky. 1161 * Although writes only happen on the CPU that they are on, 1162 * and they only need to worry about interrupts. Reads can 1163 * happen on any CPU. 1164 * 1165 * The reader page is always off the ring buffer, but when the 1166 * reader finishes with a page, it needs to swap its page with 1167 * a new one from the buffer. The reader needs to take from 1168 * the head (writes go to the tail). But if a writer is in overwrite 1169 * mode and wraps, it must push the head page forward. 1170 * 1171 * Here lies the problem. 1172 * 1173 * The reader must be careful to replace only the head page, and 1174 * not another one. As described at the top of the file in the 1175 * ASCII art, the reader sets its old page to point to the next 1176 * page after head. It then sets the page after head to point to 1177 * the old reader page. But if the writer moves the head page 1178 * during this operation, the reader could end up with the tail. 1179 * 1180 * We use cmpxchg to help prevent this race. We also do something 1181 * special with the page before head. We set the LSB to 1. 1182 * 1183 * When the writer must push the page forward, it will clear the 1184 * bit that points to the head page, move the head, and then set 1185 * the bit that points to the new head page. 1186 * 1187 * We also don't want an interrupt coming in and moving the head 1188 * page on another writer. Thus we use the second LSB to catch 1189 * that too. Thus: 1190 * 1191 * head->list->prev->next bit 1 bit 0 1192 * ------- ------- 1193 * Normal page 0 0 1194 * Points to head page 0 1 1195 * New head page 1 0 1196 * 1197 * Note we can not trust the prev pointer of the head page, because: 1198 * 1199 * +----+ +-----+ +-----+ 1200 * | |------>| T |---X--->| N | 1201 * | |<------| | | | 1202 * +----+ +-----+ +-----+ 1203 * ^ ^ | 1204 * | +-----+ | | 1205 * +----------| R |----------+ | 1206 * | |<-----------+ 1207 * +-----+ 1208 * 1209 * Key: ---X--> HEAD flag set in pointer 1210 * T Tail page 1211 * R Reader page 1212 * N Next page 1213 * 1214 * (see __rb_reserve_next() to see where this happens) 1215 * 1216 * What the above shows is that the reader just swapped out 1217 * the reader page with a page in the buffer, but before it 1218 * could make the new header point back to the new page added 1219 * it was preempted by a writer. The writer moved forward onto 1220 * the new page added by the reader and is about to move forward 1221 * again. 1222 * 1223 * You can see, it is legitimate for the previous pointer of 1224 * the head (or any page) not to point back to itself. But only 1225 * temporarily. 1226 */ 1227 1228#define RB_PAGE_NORMAL 0UL 1229#define RB_PAGE_HEAD 1UL 1230#define RB_PAGE_UPDATE 2UL 1231 1232 1233#define RB_FLAG_MASK 3UL 1234 1235/* PAGE_MOVED is not part of the mask */ 1236#define RB_PAGE_MOVED 4UL 1237 1238/* 1239 * rb_list_head - remove any bit 1240 */ 1241static struct list_head *rb_list_head(struct list_head *list) 1242{ 1243 unsigned long val = (unsigned long)list; 1244 1245 return (struct list_head *)(val & ~RB_FLAG_MASK); 1246} 1247 1248/* 1249 * rb_is_head_page - test if the given page is the head page 1250 * 1251 * Because the reader may move the head_page pointer, we can 1252 * not trust what the head page is (it may be pointing to 1253 * the reader page). But if the next page is a header page, 1254 * its flags will be non zero. 1255 */ 1256static inline int 1257rb_is_head_page(struct buffer_page *page, struct list_head *list) 1258{ 1259 unsigned long val; 1260 1261 val = (unsigned long)list->next; 1262 1263 if ((val & ~RB_FLAG_MASK) != (unsigned long)&page->list) 1264 return RB_PAGE_MOVED; 1265 1266 return val & RB_FLAG_MASK; 1267} 1268 1269/* 1270 * rb_is_reader_page 1271 * 1272 * The unique thing about the reader page, is that, if the 1273 * writer is ever on it, the previous pointer never points 1274 * back to the reader page. 1275 */ 1276static bool rb_is_reader_page(struct buffer_page *page) 1277{ 1278 struct list_head *list = page->list.prev; 1279 1280 return rb_list_head(list->next) != &page->list; 1281} 1282 1283/* 1284 * rb_set_list_to_head - set a list_head to be pointing to head. 1285 */ 1286static void rb_set_list_to_head(struct list_head *list) 1287{ 1288 unsigned long *ptr; 1289 1290 ptr = (unsigned long *)&list->next; 1291 *ptr |= RB_PAGE_HEAD; 1292 *ptr &= ~RB_PAGE_UPDATE; 1293} 1294 1295/* 1296 * rb_head_page_activate - sets up head page 1297 */ 1298static void rb_head_page_activate(struct ring_buffer_per_cpu *cpu_buffer) 1299{ 1300 struct buffer_page *head; 1301 1302 head = cpu_buffer->head_page; 1303 if (!head) 1304 return; 1305 1306 /* 1307 * Set the previous list pointer to have the HEAD flag. 1308 */ 1309 rb_set_list_to_head(head->list.prev); 1310 1311 if (cpu_buffer->ring_meta) { 1312 struct ring_buffer_cpu_meta *meta = cpu_buffer->ring_meta; 1313 meta->head_buffer = (unsigned long)head->page; 1314 } 1315} 1316 1317static void rb_list_head_clear(struct list_head *list) 1318{ 1319 unsigned long *ptr = (unsigned long *)&list->next; 1320 1321 *ptr &= ~RB_FLAG_MASK; 1322} 1323 1324/* 1325 * rb_head_page_deactivate - clears head page ptr (for free list) 1326 */ 1327static void 1328rb_head_page_deactivate(struct ring_buffer_per_cpu *cpu_buffer) 1329{ 1330 struct list_head *hd; 1331 1332 /* Go through the whole list and clear any pointers found. */ 1333 rb_list_head_clear(cpu_buffer->pages); 1334 1335 list_for_each(hd, cpu_buffer->pages) 1336 rb_list_head_clear(hd); 1337} 1338 1339static int rb_head_page_set(struct ring_buffer_per_cpu *cpu_buffer, 1340 struct buffer_page *head, 1341 struct buffer_page *prev, 1342 int old_flag, int new_flag) 1343{ 1344 struct list_head *list; 1345 unsigned long val = (unsigned long)&head->list; 1346 unsigned long ret; 1347 1348 list = &prev->list; 1349 1350 val &= ~RB_FLAG_MASK; 1351 1352 ret = cmpxchg((unsigned long *)&list->next, 1353 val | old_flag, val | new_flag); 1354 1355 /* check if the reader took the page */ 1356 if ((ret & ~RB_FLAG_MASK) != val) 1357 return RB_PAGE_MOVED; 1358 1359 return ret & RB_FLAG_MASK; 1360} 1361 1362static int rb_head_page_set_update(struct ring_buffer_per_cpu *cpu_buffer, 1363 struct buffer_page *head, 1364 struct buffer_page *prev, 1365 int old_flag) 1366{ 1367 return rb_head_page_set(cpu_buffer, head, prev, 1368 old_flag, RB_PAGE_UPDATE); 1369} 1370 1371static int rb_head_page_set_head(struct ring_buffer_per_cpu *cpu_buffer, 1372 struct buffer_page *head, 1373 struct buffer_page *prev, 1374 int old_flag) 1375{ 1376 return rb_head_page_set(cpu_buffer, head, prev, 1377 old_flag, RB_PAGE_HEAD); 1378} 1379 1380static int rb_head_page_set_normal(struct ring_buffer_per_cpu *cpu_buffer, 1381 struct buffer_page *head, 1382 struct buffer_page *prev, 1383 int old_flag) 1384{ 1385 return rb_head_page_set(cpu_buffer, head, prev, 1386 old_flag, RB_PAGE_NORMAL); 1387} 1388 1389static inline void rb_inc_page(struct buffer_page **bpage) 1390{ 1391 struct list_head *p = rb_list_head((*bpage)->list.next); 1392 1393 *bpage = list_entry(p, struct buffer_page, list); 1394} 1395 1396static inline void rb_dec_page(struct buffer_page **bpage) 1397{ 1398 struct list_head *p = rb_list_head((*bpage)->list.prev); 1399 1400 *bpage = list_entry(p, struct buffer_page, list); 1401} 1402 1403static struct buffer_page * 1404rb_set_head_page(struct ring_buffer_per_cpu *cpu_buffer) 1405{ 1406 struct buffer_page *head; 1407 struct buffer_page *page; 1408 struct list_head *list; 1409 int i; 1410 1411 if (RB_WARN_ON(cpu_buffer, !cpu_buffer->head_page)) 1412 return NULL; 1413 1414 /* sanity check */ 1415 list = cpu_buffer->pages; 1416 if (RB_WARN_ON(cpu_buffer, rb_list_head(list->prev->next) != list)) 1417 return NULL; 1418 1419 page = head = cpu_buffer->head_page; 1420 /* 1421 * It is possible that the writer moves the header behind 1422 * where we started, and we miss in one loop. 1423 * A second loop should grab the header, but we'll do 1424 * three loops just because I'm paranoid. 1425 */ 1426 for (i = 0; i < 3; i++) { 1427 do { 1428 if (rb_is_head_page(page, page->list.prev)) { 1429 cpu_buffer->head_page = page; 1430 return page; 1431 } 1432 rb_inc_page(&page); 1433 } while (page != head); 1434 } 1435 1436 RB_WARN_ON(cpu_buffer, 1); 1437 1438 return NULL; 1439} 1440 1441static bool rb_head_page_replace(struct buffer_page *old, 1442 struct buffer_page *new) 1443{ 1444 unsigned long *ptr = (unsigned long *)&old->list.prev->next; 1445 unsigned long val; 1446 1447 val = *ptr & ~RB_FLAG_MASK; 1448 val |= RB_PAGE_HEAD; 1449 1450 return try_cmpxchg(ptr, &val, (unsigned long)&new->list); 1451} 1452 1453/* 1454 * rb_tail_page_update - move the tail page forward 1455 */ 1456static void rb_tail_page_update(struct ring_buffer_per_cpu *cpu_buffer, 1457 struct buffer_page *tail_page, 1458 struct buffer_page *next_page) 1459{ 1460 unsigned long old_entries; 1461 unsigned long old_write; 1462 1463 /* 1464 * The tail page now needs to be moved forward. 1465 * 1466 * We need to reset the tail page, but without messing 1467 * with possible erasing of data brought in by interrupts 1468 * that have moved the tail page and are currently on it. 1469 * 1470 * We add a counter to the write field to denote this. 1471 */ 1472 old_write = local_add_return(RB_WRITE_INTCNT, &next_page->write); 1473 old_entries = local_add_return(RB_WRITE_INTCNT, &next_page->entries); 1474 1475 /* 1476 * Just make sure we have seen our old_write and synchronize 1477 * with any interrupts that come in. 1478 */ 1479 barrier(); 1480 1481 /* 1482 * If the tail page is still the same as what we think 1483 * it is, then it is up to us to update the tail 1484 * pointer. 1485 */ 1486 if (tail_page == READ_ONCE(cpu_buffer->tail_page)) { 1487 /* Zero the write counter */ 1488 unsigned long val = old_write & ~RB_WRITE_MASK; 1489 unsigned long eval = old_entries & ~RB_WRITE_MASK; 1490 1491 /* 1492 * This will only succeed if an interrupt did 1493 * not come in and change it. In which case, we 1494 * do not want to modify it. 1495 * 1496 * We add (void) to let the compiler know that we do not care 1497 * about the return value of these functions. We use the 1498 * cmpxchg to only update if an interrupt did not already 1499 * do it for us. If the cmpxchg fails, we don't care. 1500 */ 1501 (void)local_cmpxchg(&next_page->write, old_write, val); 1502 (void)local_cmpxchg(&next_page->entries, old_entries, eval); 1503 1504 /* 1505 * No need to worry about races with clearing out the commit. 1506 * it only can increment when a commit takes place. But that 1507 * only happens in the outer most nested commit. 1508 */ 1509 local_set(&next_page->page->commit, 0); 1510 1511 /* Either we update tail_page or an interrupt does */ 1512 if (try_cmpxchg(&cpu_buffer->tail_page, &tail_page, next_page)) 1513 local_inc(&cpu_buffer->pages_touched); 1514 } 1515} 1516 1517static void rb_check_bpage(struct ring_buffer_per_cpu *cpu_buffer, 1518 struct buffer_page *bpage) 1519{ 1520 unsigned long val = (unsigned long)bpage; 1521 1522 RB_WARN_ON(cpu_buffer, val & RB_FLAG_MASK); 1523} 1524 1525static bool rb_check_links(struct ring_buffer_per_cpu *cpu_buffer, 1526 struct list_head *list) 1527{ 1528 if (RB_WARN_ON(cpu_buffer, 1529 rb_list_head(rb_list_head(list->next)->prev) != list)) 1530 return false; 1531 1532 if (RB_WARN_ON(cpu_buffer, 1533 rb_list_head(rb_list_head(list->prev)->next) != list)) 1534 return false; 1535 1536 return true; 1537} 1538 1539/** 1540 * rb_check_pages - integrity check of buffer pages 1541 * @cpu_buffer: CPU buffer with pages to test 1542 * 1543 * As a safety measure we check to make sure the data pages have not 1544 * been corrupted. 1545 */ 1546static void rb_check_pages(struct ring_buffer_per_cpu *cpu_buffer) 1547{ 1548 struct list_head *head, *tmp; 1549 unsigned long buffer_cnt; 1550 unsigned long flags; 1551 int nr_loops = 0; 1552 1553 /* 1554 * Walk the linked list underpinning the ring buffer and validate all 1555 * its next and prev links. 1556 * 1557 * The check acquires the reader_lock to avoid concurrent processing 1558 * with code that could be modifying the list. However, the lock cannot 1559 * be held for the entire duration of the walk, as this would make the 1560 * time when interrupts are disabled non-deterministic, dependent on the 1561 * ring buffer size. Therefore, the code releases and re-acquires the 1562 * lock after checking each page. The ring_buffer_per_cpu.cnt variable 1563 * is then used to detect if the list was modified while the lock was 1564 * not held, in which case the check needs to be restarted. 1565 * 1566 * The code attempts to perform the check at most three times before 1567 * giving up. This is acceptable because this is only a self-validation 1568 * to detect problems early on. In practice, the list modification 1569 * operations are fairly spaced, and so this check typically succeeds at 1570 * most on the second try. 1571 */ 1572again: 1573 if (++nr_loops > 3) 1574 return; 1575 1576 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 1577 head = rb_list_head(cpu_buffer->pages); 1578 if (!rb_check_links(cpu_buffer, head)) 1579 goto out_locked; 1580 buffer_cnt = cpu_buffer->cnt; 1581 tmp = head; 1582 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 1583 1584 while (true) { 1585 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 1586 1587 if (buffer_cnt != cpu_buffer->cnt) { 1588 /* The list was updated, try again. */ 1589 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 1590 goto again; 1591 } 1592 1593 tmp = rb_list_head(tmp->next); 1594 if (tmp == head) 1595 /* The iteration circled back, all is done. */ 1596 goto out_locked; 1597 1598 if (!rb_check_links(cpu_buffer, tmp)) 1599 goto out_locked; 1600 1601 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 1602 } 1603 1604out_locked: 1605 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 1606} 1607 1608/* 1609 * Take an address, add the meta data size as well as the array of 1610 * array subbuffer indexes, then align it to a subbuffer size. 1611 * 1612 * This is used to help find the next per cpu subbuffer within a mapped range. 1613 */ 1614static unsigned long 1615rb_range_align_subbuf(unsigned long addr, int subbuf_size, int nr_subbufs) 1616{ 1617 addr += sizeof(struct ring_buffer_cpu_meta) + 1618 sizeof(int) * nr_subbufs; 1619 return ALIGN(addr, subbuf_size); 1620} 1621 1622/* 1623 * Return the ring_buffer_meta for a given @cpu. 1624 */ 1625static void *rb_range_meta(struct trace_buffer *buffer, int nr_pages, int cpu) 1626{ 1627 int subbuf_size = buffer->subbuf_size + BUF_PAGE_HDR_SIZE; 1628 struct ring_buffer_cpu_meta *meta; 1629 struct ring_buffer_meta *bmeta; 1630 unsigned long ptr; 1631 int nr_subbufs; 1632 1633 bmeta = buffer->meta; 1634 if (!bmeta) 1635 return NULL; 1636 1637 ptr = (unsigned long)bmeta + bmeta->buffers_offset; 1638 meta = (struct ring_buffer_cpu_meta *)ptr; 1639 1640 /* When nr_pages passed in is zero, the first meta has already been initialized */ 1641 if (!nr_pages) { 1642 nr_subbufs = meta->nr_subbufs; 1643 } else { 1644 /* Include the reader page */ 1645 nr_subbufs = nr_pages + 1; 1646 } 1647 1648 /* 1649 * The first chunk may not be subbuffer aligned, where as 1650 * the rest of the chunks are. 1651 */ 1652 if (cpu) { 1653 ptr = rb_range_align_subbuf(ptr, subbuf_size, nr_subbufs); 1654 ptr += subbuf_size * nr_subbufs; 1655 1656 /* We can use multiplication to find chunks greater than 1 */ 1657 if (cpu > 1) { 1658 unsigned long size; 1659 unsigned long p; 1660 1661 /* Save the beginning of this CPU chunk */ 1662 p = ptr; 1663 ptr = rb_range_align_subbuf(ptr, subbuf_size, nr_subbufs); 1664 ptr += subbuf_size * nr_subbufs; 1665 1666 /* Now all chunks after this are the same size */ 1667 size = ptr - p; 1668 ptr += size * (cpu - 2); 1669 } 1670 } 1671 return (void *)ptr; 1672} 1673 1674/* Return the start of subbufs given the meta pointer */ 1675static void *rb_subbufs_from_meta(struct ring_buffer_cpu_meta *meta) 1676{ 1677 int subbuf_size = meta->subbuf_size; 1678 unsigned long ptr; 1679 1680 ptr = (unsigned long)meta; 1681 ptr = rb_range_align_subbuf(ptr, subbuf_size, meta->nr_subbufs); 1682 1683 return (void *)ptr; 1684} 1685 1686/* 1687 * Return a specific sub-buffer for a given @cpu defined by @idx. 1688 */ 1689static void *rb_range_buffer(struct ring_buffer_per_cpu *cpu_buffer, int idx) 1690{ 1691 struct ring_buffer_cpu_meta *meta; 1692 unsigned long ptr; 1693 int subbuf_size; 1694 1695 meta = rb_range_meta(cpu_buffer->buffer, 0, cpu_buffer->cpu); 1696 if (!meta) 1697 return NULL; 1698 1699 if (WARN_ON_ONCE(idx >= meta->nr_subbufs)) 1700 return NULL; 1701 1702 subbuf_size = meta->subbuf_size; 1703 1704 /* Map this buffer to the order that's in meta->buffers[] */ 1705 idx = meta->buffers[idx]; 1706 1707 ptr = (unsigned long)rb_subbufs_from_meta(meta); 1708 1709 ptr += subbuf_size * idx; 1710 if (ptr + subbuf_size > cpu_buffer->buffer->range_addr_end) 1711 return NULL; 1712 1713 return (void *)ptr; 1714} 1715 1716/* 1717 * See if the existing memory contains a valid meta section. 1718 * if so, use that, otherwise initialize it. 1719 */ 1720static bool rb_meta_init(struct trace_buffer *buffer, int scratch_size) 1721{ 1722 unsigned long ptr = buffer->range_addr_start; 1723 struct ring_buffer_meta *bmeta; 1724 unsigned long total_size; 1725 int struct_sizes; 1726 1727 bmeta = (struct ring_buffer_meta *)ptr; 1728 buffer->meta = bmeta; 1729 1730 total_size = buffer->range_addr_end - buffer->range_addr_start; 1731 1732 struct_sizes = sizeof(struct ring_buffer_cpu_meta); 1733 struct_sizes |= sizeof(*bmeta) << 16; 1734 1735 /* The first buffer will start word size after the meta page */ 1736 ptr += sizeof(*bmeta); 1737 ptr = ALIGN(ptr, sizeof(long)); 1738 ptr += scratch_size; 1739 1740 if (bmeta->magic != RING_BUFFER_META_MAGIC) { 1741 pr_info("Ring buffer boot meta mismatch of magic\n"); 1742 goto init; 1743 } 1744 1745 if (bmeta->struct_sizes != struct_sizes) { 1746 pr_info("Ring buffer boot meta mismatch of struct size\n"); 1747 goto init; 1748 } 1749 1750 if (bmeta->total_size != total_size) { 1751 pr_info("Ring buffer boot meta mismatch of total size\n"); 1752 goto init; 1753 } 1754 1755 if (bmeta->buffers_offset > bmeta->total_size) { 1756 pr_info("Ring buffer boot meta mismatch of offset outside of total size\n"); 1757 goto init; 1758 } 1759 1760 if (bmeta->buffers_offset != (void *)ptr - (void *)bmeta) { 1761 pr_info("Ring buffer boot meta mismatch of first buffer offset\n"); 1762 goto init; 1763 } 1764 1765 return true; 1766 1767 init: 1768 bmeta->magic = RING_BUFFER_META_MAGIC; 1769 bmeta->struct_sizes = struct_sizes; 1770 bmeta->total_size = total_size; 1771 bmeta->buffers_offset = (void *)ptr - (void *)bmeta; 1772 1773 /* Zero out the scratch pad */ 1774 memset((void *)bmeta + sizeof(*bmeta), 0, bmeta->buffers_offset - sizeof(*bmeta)); 1775 1776 return false; 1777} 1778 1779/* 1780 * See if the existing memory contains valid ring buffer data. 1781 * As the previous kernel must be the same as this kernel, all 1782 * the calculations (size of buffers and number of buffers) 1783 * must be the same. 1784 */ 1785static bool rb_cpu_meta_valid(struct ring_buffer_cpu_meta *meta, int cpu, 1786 struct trace_buffer *buffer, int nr_pages, 1787 unsigned long *subbuf_mask) 1788{ 1789 int subbuf_size = PAGE_SIZE; 1790 struct buffer_data_page *subbuf; 1791 unsigned long buffers_start; 1792 unsigned long buffers_end; 1793 int i; 1794 1795 if (!subbuf_mask) 1796 return false; 1797 1798 buffers_start = meta->first_buffer; 1799 buffers_end = meta->first_buffer + (subbuf_size * meta->nr_subbufs); 1800 1801 /* Is the head and commit buffers within the range of buffers? */ 1802 if (meta->head_buffer < buffers_start || 1803 meta->head_buffer >= buffers_end) { 1804 pr_info("Ring buffer boot meta [%d] head buffer out of range\n", cpu); 1805 return false; 1806 } 1807 1808 if (meta->commit_buffer < buffers_start || 1809 meta->commit_buffer >= buffers_end) { 1810 pr_info("Ring buffer boot meta [%d] commit buffer out of range\n", cpu); 1811 return false; 1812 } 1813 1814 subbuf = rb_subbufs_from_meta(meta); 1815 1816 bitmap_clear(subbuf_mask, 0, meta->nr_subbufs); 1817 1818 /* Is the meta buffers and the subbufs themselves have correct data? */ 1819 for (i = 0; i < meta->nr_subbufs; i++) { 1820 if (meta->buffers[i] < 0 || 1821 meta->buffers[i] >= meta->nr_subbufs) { 1822 pr_info("Ring buffer boot meta [%d] array out of range\n", cpu); 1823 return false; 1824 } 1825 1826 if ((unsigned)local_read(&subbuf->commit) > subbuf_size) { 1827 pr_info("Ring buffer boot meta [%d] buffer invalid commit\n", cpu); 1828 return false; 1829 } 1830 1831 if (test_bit(meta->buffers[i], subbuf_mask)) { 1832 pr_info("Ring buffer boot meta [%d] array has duplicates\n", cpu); 1833 return false; 1834 } 1835 1836 set_bit(meta->buffers[i], subbuf_mask); 1837 subbuf = (void *)subbuf + subbuf_size; 1838 } 1839 1840 return true; 1841} 1842 1843static int rb_meta_subbuf_idx(struct ring_buffer_cpu_meta *meta, void *subbuf); 1844 1845static int rb_read_data_buffer(struct buffer_data_page *dpage, int tail, int cpu, 1846 unsigned long long *timestamp, u64 *delta_ptr) 1847{ 1848 struct ring_buffer_event *event; 1849 u64 ts, delta; 1850 int events = 0; 1851 int len; 1852 int e; 1853 1854 *delta_ptr = 0; 1855 *timestamp = 0; 1856 1857 ts = dpage->time_stamp; 1858 1859 for (e = 0; e < tail; e += len) { 1860 1861 event = (struct ring_buffer_event *)(dpage->data + e); 1862 len = rb_event_length(event); 1863 if (len <= 0 || len > tail - e) 1864 return -1; 1865 1866 switch (event->type_len) { 1867 1868 case RINGBUF_TYPE_TIME_EXTEND: 1869 delta = rb_event_time_stamp(event); 1870 ts += delta; 1871 break; 1872 1873 case RINGBUF_TYPE_TIME_STAMP: 1874 delta = rb_event_time_stamp(event); 1875 delta = rb_fix_abs_ts(delta, ts); 1876 if (delta < ts) { 1877 *delta_ptr = delta; 1878 *timestamp = ts; 1879 return -1; 1880 } 1881 ts = delta; 1882 break; 1883 1884 case RINGBUF_TYPE_PADDING: 1885 if (event->time_delta == 1) 1886 break; 1887 fallthrough; 1888 case RINGBUF_TYPE_DATA: 1889 events++; 1890 ts += event->time_delta; 1891 break; 1892 1893 default: 1894 return -1; 1895 } 1896 } 1897 *timestamp = ts; 1898 return events; 1899} 1900 1901static int rb_validate_buffer(struct buffer_data_page *dpage, int cpu) 1902{ 1903 unsigned long long ts; 1904 u64 delta; 1905 int tail; 1906 1907 tail = local_read(&dpage->commit); 1908 return rb_read_data_buffer(dpage, tail, cpu, &ts, &delta); 1909} 1910 1911/* If the meta data has been validated, now validate the events */ 1912static void rb_meta_validate_events(struct ring_buffer_per_cpu *cpu_buffer) 1913{ 1914 struct ring_buffer_cpu_meta *meta = cpu_buffer->ring_meta; 1915 struct buffer_page *head_page, *orig_head; 1916 unsigned long entry_bytes = 0; 1917 unsigned long entries = 0; 1918 int ret; 1919 u64 ts; 1920 int i; 1921 1922 if (!meta || !meta->head_buffer) 1923 return; 1924 1925 orig_head = head_page = cpu_buffer->head_page; 1926 1927 /* Do the reader page first */ 1928 ret = rb_validate_buffer(cpu_buffer->reader_page->page, cpu_buffer->cpu); 1929 if (ret < 0) { 1930 pr_info("Ring buffer reader page is invalid\n"); 1931 goto invalid; 1932 } 1933 entries += ret; 1934 entry_bytes += local_read(&cpu_buffer->reader_page->page->commit); 1935 local_set(&cpu_buffer->reader_page->entries, ret); 1936 1937 ts = head_page->page->time_stamp; 1938 1939 /* 1940 * Try to rewind the head so that we can read the pages which already 1941 * read in the previous boot. 1942 */ 1943 if (head_page == cpu_buffer->tail_page) 1944 goto skip_rewind; 1945 1946 rb_dec_page(&head_page); 1947 for (i = 0; i < meta->nr_subbufs + 1; i++, rb_dec_page(&head_page)) { 1948 1949 /* Rewind until tail (writer) page. */ 1950 if (head_page == cpu_buffer->tail_page) 1951 break; 1952 1953 /* Ensure the page has older data than head. */ 1954 if (ts < head_page->page->time_stamp) 1955 break; 1956 1957 ts = head_page->page->time_stamp; 1958 /* Ensure the page has correct timestamp and some data. */ 1959 if (!ts || rb_page_commit(head_page) == 0) 1960 break; 1961 1962 /* Stop rewind if the page is invalid. */ 1963 ret = rb_validate_buffer(head_page->page, cpu_buffer->cpu); 1964 if (ret < 0) 1965 break; 1966 1967 /* Recover the number of entries and update stats. */ 1968 local_set(&head_page->entries, ret); 1969 if (ret) 1970 local_inc(&cpu_buffer->pages_touched); 1971 entries += ret; 1972 entry_bytes += rb_page_commit(head_page); 1973 } 1974 if (i) 1975 pr_info("Ring buffer [%d] rewound %d pages\n", cpu_buffer->cpu, i); 1976 1977 /* The last rewound page must be skipped. */ 1978 if (head_page != orig_head) 1979 rb_inc_page(&head_page); 1980 1981 /* 1982 * If the ring buffer was rewound, then inject the reader page 1983 * into the location just before the original head page. 1984 */ 1985 if (head_page != orig_head) { 1986 struct buffer_page *bpage = orig_head; 1987 1988 rb_dec_page(&bpage); 1989 /* 1990 * Insert the reader_page before the original head page. 1991 * Since the list encode RB_PAGE flags, general list 1992 * operations should be avoided. 1993 */ 1994 cpu_buffer->reader_page->list.next = &orig_head->list; 1995 cpu_buffer->reader_page->list.prev = orig_head->list.prev; 1996 orig_head->list.prev = &cpu_buffer->reader_page->list; 1997 bpage->list.next = &cpu_buffer->reader_page->list; 1998 1999 /* Make the head_page the reader page */ 2000 cpu_buffer->reader_page = head_page; 2001 bpage = head_page; 2002 rb_inc_page(&head_page); 2003 head_page->list.prev = bpage->list.prev; 2004 rb_dec_page(&bpage); 2005 bpage->list.next = &head_page->list; 2006 rb_set_list_to_head(&bpage->list); 2007 cpu_buffer->pages = &head_page->list; 2008 2009 cpu_buffer->head_page = head_page; 2010 meta->head_buffer = (unsigned long)head_page->page; 2011 2012 /* Reset all the indexes */ 2013 bpage = cpu_buffer->reader_page; 2014 meta->buffers[0] = rb_meta_subbuf_idx(meta, bpage->page); 2015 bpage->id = 0; 2016 2017 for (i = 1, bpage = head_page; i < meta->nr_subbufs; 2018 i++, rb_inc_page(&bpage)) { 2019 meta->buffers[i] = rb_meta_subbuf_idx(meta, bpage->page); 2020 bpage->id = i; 2021 } 2022 2023 /* We'll restart verifying from orig_head */ 2024 head_page = orig_head; 2025 } 2026 2027 skip_rewind: 2028 /* If the commit_buffer is the reader page, update the commit page */ 2029 if (meta->commit_buffer == (unsigned long)cpu_buffer->reader_page->page) { 2030 cpu_buffer->commit_page = cpu_buffer->reader_page; 2031 /* Nothing more to do, the only page is the reader page */ 2032 goto done; 2033 } 2034 2035 /* Iterate until finding the commit page */ 2036 for (i = 0; i < meta->nr_subbufs + 1; i++, rb_inc_page(&head_page)) { 2037 2038 /* Reader page has already been done */ 2039 if (head_page == cpu_buffer->reader_page) 2040 continue; 2041 2042 ret = rb_validate_buffer(head_page->page, cpu_buffer->cpu); 2043 if (ret < 0) { 2044 pr_info("Ring buffer meta [%d] invalid buffer page\n", 2045 cpu_buffer->cpu); 2046 goto invalid; 2047 } 2048 2049 /* If the buffer has content, update pages_touched */ 2050 if (ret) 2051 local_inc(&cpu_buffer->pages_touched); 2052 2053 entries += ret; 2054 entry_bytes += local_read(&head_page->page->commit); 2055 local_set(&head_page->entries, ret); 2056 2057 if (head_page == cpu_buffer->commit_page) 2058 break; 2059 } 2060 2061 if (head_page != cpu_buffer->commit_page) { 2062 pr_info("Ring buffer meta [%d] commit page not found\n", 2063 cpu_buffer->cpu); 2064 goto invalid; 2065 } 2066 done: 2067 local_set(&cpu_buffer->entries, entries); 2068 local_set(&cpu_buffer->entries_bytes, entry_bytes); 2069 2070 pr_info("Ring buffer meta [%d] is from previous boot!\n", cpu_buffer->cpu); 2071 return; 2072 2073 invalid: 2074 /* The content of the buffers are invalid, reset the meta data */ 2075 meta->head_buffer = 0; 2076 meta->commit_buffer = 0; 2077 2078 /* Reset the reader page */ 2079 local_set(&cpu_buffer->reader_page->entries, 0); 2080 local_set(&cpu_buffer->reader_page->page->commit, 0); 2081 2082 /* Reset all the subbuffers */ 2083 for (i = 0; i < meta->nr_subbufs - 1; i++, rb_inc_page(&head_page)) { 2084 local_set(&head_page->entries, 0); 2085 local_set(&head_page->page->commit, 0); 2086 } 2087} 2088 2089static void rb_range_meta_init(struct trace_buffer *buffer, int nr_pages, int scratch_size) 2090{ 2091 struct ring_buffer_cpu_meta *meta; 2092 unsigned long *subbuf_mask; 2093 unsigned long delta; 2094 void *subbuf; 2095 bool valid = false; 2096 int cpu; 2097 int i; 2098 2099 /* Create a mask to test the subbuf array */ 2100 subbuf_mask = bitmap_alloc(nr_pages + 1, GFP_KERNEL); 2101 /* If subbuf_mask fails to allocate, then rb_meta_valid() will return false */ 2102 2103 if (rb_meta_init(buffer, scratch_size)) 2104 valid = true; 2105 2106 for (cpu = 0; cpu < nr_cpu_ids; cpu++) { 2107 void *next_meta; 2108 2109 meta = rb_range_meta(buffer, nr_pages, cpu); 2110 2111 if (valid && rb_cpu_meta_valid(meta, cpu, buffer, nr_pages, subbuf_mask)) { 2112 /* Make the mappings match the current address */ 2113 subbuf = rb_subbufs_from_meta(meta); 2114 delta = (unsigned long)subbuf - meta->first_buffer; 2115 meta->first_buffer += delta; 2116 meta->head_buffer += delta; 2117 meta->commit_buffer += delta; 2118 continue; 2119 } 2120 2121 if (cpu < nr_cpu_ids - 1) 2122 next_meta = rb_range_meta(buffer, nr_pages, cpu + 1); 2123 else 2124 next_meta = (void *)buffer->range_addr_end; 2125 2126 memset(meta, 0, next_meta - (void *)meta); 2127 2128 meta->nr_subbufs = nr_pages + 1; 2129 meta->subbuf_size = PAGE_SIZE; 2130 2131 subbuf = rb_subbufs_from_meta(meta); 2132 2133 meta->first_buffer = (unsigned long)subbuf; 2134 2135 /* 2136 * The buffers[] array holds the order of the sub-buffers 2137 * that are after the meta data. The sub-buffers may 2138 * be swapped out when read and inserted into a different 2139 * location of the ring buffer. Although their addresses 2140 * remain the same, the buffers[] array contains the 2141 * index into the sub-buffers holding their actual order. 2142 */ 2143 for (i = 0; i < meta->nr_subbufs; i++) { 2144 meta->buffers[i] = i; 2145 rb_init_page(subbuf); 2146 subbuf += meta->subbuf_size; 2147 } 2148 } 2149 bitmap_free(subbuf_mask); 2150} 2151 2152static void *rbm_start(struct seq_file *m, loff_t *pos) 2153{ 2154 struct ring_buffer_per_cpu *cpu_buffer = m->private; 2155 struct ring_buffer_cpu_meta *meta = cpu_buffer->ring_meta; 2156 unsigned long val; 2157 2158 if (!meta) 2159 return NULL; 2160 2161 if (*pos > meta->nr_subbufs) 2162 return NULL; 2163 2164 val = *pos; 2165 val++; 2166 2167 return (void *)val; 2168} 2169 2170static void *rbm_next(struct seq_file *m, void *v, loff_t *pos) 2171{ 2172 (*pos)++; 2173 2174 return rbm_start(m, pos); 2175} 2176 2177static int rbm_show(struct seq_file *m, void *v) 2178{ 2179 struct ring_buffer_per_cpu *cpu_buffer = m->private; 2180 struct ring_buffer_cpu_meta *meta = cpu_buffer->ring_meta; 2181 unsigned long val = (unsigned long)v; 2182 2183 if (val == 1) { 2184 seq_printf(m, "head_buffer: %d\n", 2185 rb_meta_subbuf_idx(meta, (void *)meta->head_buffer)); 2186 seq_printf(m, "commit_buffer: %d\n", 2187 rb_meta_subbuf_idx(meta, (void *)meta->commit_buffer)); 2188 seq_printf(m, "subbuf_size: %d\n", meta->subbuf_size); 2189 seq_printf(m, "nr_subbufs: %d\n", meta->nr_subbufs); 2190 return 0; 2191 } 2192 2193 val -= 2; 2194 seq_printf(m, "buffer[%ld]: %d\n", val, meta->buffers[val]); 2195 2196 return 0; 2197} 2198 2199static void rbm_stop(struct seq_file *m, void *p) 2200{ 2201} 2202 2203static const struct seq_operations rb_meta_seq_ops = { 2204 .start = rbm_start, 2205 .next = rbm_next, 2206 .show = rbm_show, 2207 .stop = rbm_stop, 2208}; 2209 2210int ring_buffer_meta_seq_init(struct file *file, struct trace_buffer *buffer, int cpu) 2211{ 2212 struct seq_file *m; 2213 int ret; 2214 2215 ret = seq_open(file, &rb_meta_seq_ops); 2216 if (ret) 2217 return ret; 2218 2219 m = file->private_data; 2220 m->private = buffer->buffers[cpu]; 2221 2222 return 0; 2223} 2224 2225/* Map the buffer_pages to the previous head and commit pages */ 2226static void rb_meta_buffer_update(struct ring_buffer_per_cpu *cpu_buffer, 2227 struct buffer_page *bpage) 2228{ 2229 struct ring_buffer_cpu_meta *meta = cpu_buffer->ring_meta; 2230 2231 if (meta->head_buffer == (unsigned long)bpage->page) 2232 cpu_buffer->head_page = bpage; 2233 2234 if (meta->commit_buffer == (unsigned long)bpage->page) { 2235 cpu_buffer->commit_page = bpage; 2236 cpu_buffer->tail_page = bpage; 2237 } 2238} 2239 2240static int __rb_allocate_pages(struct ring_buffer_per_cpu *cpu_buffer, 2241 long nr_pages, struct list_head *pages) 2242{ 2243 struct trace_buffer *buffer = cpu_buffer->buffer; 2244 struct ring_buffer_cpu_meta *meta = NULL; 2245 struct buffer_page *bpage, *tmp; 2246 bool user_thread = current->mm != NULL; 2247 long i; 2248 2249 /* 2250 * Check if the available memory is there first. 2251 * Note, si_mem_available() only gives us a rough estimate of available 2252 * memory. It may not be accurate. But we don't care, we just want 2253 * to prevent doing any allocation when it is obvious that it is 2254 * not going to succeed. 2255 */ 2256 i = si_mem_available(); 2257 if (i < nr_pages) 2258 return -ENOMEM; 2259 2260 /* 2261 * If a user thread allocates too much, and si_mem_available() 2262 * reports there's enough memory, even though there is not. 2263 * Make sure the OOM killer kills this thread. This can happen 2264 * even with RETRY_MAYFAIL because another task may be doing 2265 * an allocation after this task has taken all memory. 2266 * This is the task the OOM killer needs to take out during this 2267 * loop, even if it was triggered by an allocation somewhere else. 2268 */ 2269 if (user_thread) 2270 set_current_oom_origin(); 2271 2272 if (buffer->range_addr_start) 2273 meta = rb_range_meta(buffer, nr_pages, cpu_buffer->cpu); 2274 2275 for (i = 0; i < nr_pages; i++) { 2276 2277 bpage = alloc_cpu_page(cpu_buffer->cpu); 2278 if (!bpage) 2279 goto free_pages; 2280 2281 rb_check_bpage(cpu_buffer, bpage); 2282 2283 /* 2284 * Append the pages as for mapped buffers we want to keep 2285 * the order 2286 */ 2287 list_add_tail(&bpage->list, pages); 2288 2289 if (meta) { 2290 /* A range was given. Use that for the buffer page */ 2291 bpage->page = rb_range_buffer(cpu_buffer, i + 1); 2292 if (!bpage->page) 2293 goto free_pages; 2294 /* If this is valid from a previous boot */ 2295 if (meta->head_buffer) 2296 rb_meta_buffer_update(cpu_buffer, bpage); 2297 bpage->range = 1; 2298 bpage->id = i + 1; 2299 } else { 2300 int order = cpu_buffer->buffer->subbuf_order; 2301 bpage->page = alloc_cpu_data(cpu_buffer->cpu, order); 2302 if (!bpage->page) 2303 goto free_pages; 2304 } 2305 bpage->order = cpu_buffer->buffer->subbuf_order; 2306 2307 if (user_thread && fatal_signal_pending(current)) 2308 goto free_pages; 2309 } 2310 if (user_thread) 2311 clear_current_oom_origin(); 2312 2313 return 0; 2314 2315free_pages: 2316 list_for_each_entry_safe(bpage, tmp, pages, list) { 2317 list_del_init(&bpage->list); 2318 free_buffer_page(bpage); 2319 } 2320 if (user_thread) 2321 clear_current_oom_origin(); 2322 2323 return -ENOMEM; 2324} 2325 2326static int rb_allocate_pages(struct ring_buffer_per_cpu *cpu_buffer, 2327 unsigned long nr_pages) 2328{ 2329 LIST_HEAD(pages); 2330 2331 WARN_ON(!nr_pages); 2332 2333 if (__rb_allocate_pages(cpu_buffer, nr_pages, &pages)) 2334 return -ENOMEM; 2335 2336 /* 2337 * The ring buffer page list is a circular list that does not 2338 * start and end with a list head. All page list items point to 2339 * other pages. 2340 */ 2341 cpu_buffer->pages = pages.next; 2342 list_del(&pages); 2343 2344 cpu_buffer->nr_pages = nr_pages; 2345 2346 rb_check_pages(cpu_buffer); 2347 2348 return 0; 2349} 2350 2351static struct ring_buffer_per_cpu * 2352rb_allocate_cpu_buffer(struct trace_buffer *buffer, long nr_pages, int cpu) 2353{ 2354 struct ring_buffer_per_cpu *cpu_buffer __free(kfree) = 2355 alloc_cpu_buffer(cpu); 2356 struct ring_buffer_cpu_meta *meta; 2357 struct buffer_page *bpage; 2358 int ret; 2359 2360 if (!cpu_buffer) 2361 return NULL; 2362 2363 cpu_buffer->cpu = cpu; 2364 cpu_buffer->buffer = buffer; 2365 raw_spin_lock_init(&cpu_buffer->reader_lock); 2366 lockdep_set_class(&cpu_buffer->reader_lock, buffer->reader_lock_key); 2367 cpu_buffer->lock = (arch_spinlock_t)__ARCH_SPIN_LOCK_UNLOCKED; 2368 INIT_WORK(&cpu_buffer->update_pages_work, update_pages_handler); 2369 init_completion(&cpu_buffer->update_done); 2370 init_irq_work(&cpu_buffer->irq_work.work, rb_wake_up_waiters); 2371 init_waitqueue_head(&cpu_buffer->irq_work.waiters); 2372 init_waitqueue_head(&cpu_buffer->irq_work.full_waiters); 2373 mutex_init(&cpu_buffer->mapping_lock); 2374 2375 bpage = alloc_cpu_page(cpu); 2376 if (!bpage) 2377 return NULL; 2378 2379 rb_check_bpage(cpu_buffer, bpage); 2380 2381 cpu_buffer->reader_page = bpage; 2382 2383 if (buffer->range_addr_start) { 2384 /* 2385 * Range mapped buffers have the same restrictions as memory 2386 * mapped ones do. 2387 */ 2388 cpu_buffer->mapped = 1; 2389 cpu_buffer->ring_meta = rb_range_meta(buffer, nr_pages, cpu); 2390 bpage->page = rb_range_buffer(cpu_buffer, 0); 2391 if (!bpage->page) 2392 goto fail_free_reader; 2393 if (cpu_buffer->ring_meta->head_buffer) 2394 rb_meta_buffer_update(cpu_buffer, bpage); 2395 bpage->range = 1; 2396 } else { 2397 int order = cpu_buffer->buffer->subbuf_order; 2398 bpage->page = alloc_cpu_data(cpu, order); 2399 if (!bpage->page) 2400 goto fail_free_reader; 2401 } 2402 2403 INIT_LIST_HEAD(&cpu_buffer->reader_page->list); 2404 INIT_LIST_HEAD(&cpu_buffer->new_pages); 2405 2406 ret = rb_allocate_pages(cpu_buffer, nr_pages); 2407 if (ret < 0) 2408 goto fail_free_reader; 2409 2410 rb_meta_validate_events(cpu_buffer); 2411 2412 /* If the boot meta was valid then this has already been updated */ 2413 meta = cpu_buffer->ring_meta; 2414 if (!meta || !meta->head_buffer || 2415 !cpu_buffer->head_page || !cpu_buffer->commit_page || !cpu_buffer->tail_page) { 2416 if (meta && meta->head_buffer && 2417 (cpu_buffer->head_page || cpu_buffer->commit_page || cpu_buffer->tail_page)) { 2418 pr_warn("Ring buffer meta buffers not all mapped\n"); 2419 if (!cpu_buffer->head_page) 2420 pr_warn(" Missing head_page\n"); 2421 if (!cpu_buffer->commit_page) 2422 pr_warn(" Missing commit_page\n"); 2423 if (!cpu_buffer->tail_page) 2424 pr_warn(" Missing tail_page\n"); 2425 } 2426 2427 cpu_buffer->head_page 2428 = list_entry(cpu_buffer->pages, struct buffer_page, list); 2429 cpu_buffer->tail_page = cpu_buffer->commit_page = cpu_buffer->head_page; 2430 2431 rb_head_page_activate(cpu_buffer); 2432 2433 if (cpu_buffer->ring_meta) 2434 meta->commit_buffer = meta->head_buffer; 2435 } else { 2436 /* The valid meta buffer still needs to activate the head page */ 2437 rb_head_page_activate(cpu_buffer); 2438 } 2439 2440 return_ptr(cpu_buffer); 2441 2442 fail_free_reader: 2443 free_buffer_page(cpu_buffer->reader_page); 2444 2445 return NULL; 2446} 2447 2448static void rb_free_cpu_buffer(struct ring_buffer_per_cpu *cpu_buffer) 2449{ 2450 struct list_head *head = cpu_buffer->pages; 2451 struct buffer_page *bpage, *tmp; 2452 2453 irq_work_sync(&cpu_buffer->irq_work.work); 2454 2455 free_buffer_page(cpu_buffer->reader_page); 2456 2457 if (head) { 2458 rb_head_page_deactivate(cpu_buffer); 2459 2460 list_for_each_entry_safe(bpage, tmp, head, list) { 2461 list_del_init(&bpage->list); 2462 free_buffer_page(bpage); 2463 } 2464 bpage = list_entry(head, struct buffer_page, list); 2465 free_buffer_page(bpage); 2466 } 2467 2468 free_page((unsigned long)cpu_buffer->free_page); 2469 2470 kfree(cpu_buffer); 2471} 2472 2473static struct trace_buffer *alloc_buffer(unsigned long size, unsigned flags, 2474 int order, unsigned long start, 2475 unsigned long end, 2476 unsigned long scratch_size, 2477 struct lock_class_key *key) 2478{ 2479 struct trace_buffer *buffer __free(kfree) = NULL; 2480 long nr_pages; 2481 int subbuf_size; 2482 int bsize; 2483 int cpu; 2484 int ret; 2485 2486 /* keep it in its own cache line */ 2487 buffer = kzalloc(ALIGN(sizeof(*buffer), cache_line_size()), 2488 GFP_KERNEL); 2489 if (!buffer) 2490 return NULL; 2491 2492 if (!zalloc_cpumask_var(&buffer->cpumask, GFP_KERNEL)) 2493 return NULL; 2494 2495 buffer->subbuf_order = order; 2496 subbuf_size = (PAGE_SIZE << order); 2497 buffer->subbuf_size = subbuf_size - BUF_PAGE_HDR_SIZE; 2498 2499 /* Max payload is buffer page size - header (8bytes) */ 2500 buffer->max_data_size = buffer->subbuf_size - (sizeof(u32) * 2); 2501 2502 buffer->flags = flags; 2503 buffer->clock = trace_clock_local; 2504 buffer->reader_lock_key = key; 2505 2506 init_irq_work(&buffer->irq_work.work, rb_wake_up_waiters); 2507 init_waitqueue_head(&buffer->irq_work.waiters); 2508 2509 buffer->cpus = nr_cpu_ids; 2510 2511 bsize = sizeof(void *) * nr_cpu_ids; 2512 buffer->buffers = kzalloc(ALIGN(bsize, cache_line_size()), 2513 GFP_KERNEL); 2514 if (!buffer->buffers) 2515 goto fail_free_cpumask; 2516 2517 /* If start/end are specified, then that overrides size */ 2518 if (start && end) { 2519 unsigned long buffers_start; 2520 unsigned long ptr; 2521 int n; 2522 2523 /* Make sure that start is word aligned */ 2524 start = ALIGN(start, sizeof(long)); 2525 2526 /* scratch_size needs to be aligned too */ 2527 scratch_size = ALIGN(scratch_size, sizeof(long)); 2528 2529 /* Subtract the buffer meta data and word aligned */ 2530 buffers_start = start + sizeof(struct ring_buffer_cpu_meta); 2531 buffers_start = ALIGN(buffers_start, sizeof(long)); 2532 buffers_start += scratch_size; 2533 2534 /* Calculate the size for the per CPU data */ 2535 size = end - buffers_start; 2536 size = size / nr_cpu_ids; 2537 2538 /* 2539 * The number of sub-buffers (nr_pages) is determined by the 2540 * total size allocated minus the meta data size. 2541 * Then that is divided by the number of per CPU buffers 2542 * needed, plus account for the integer array index that 2543 * will be appended to the meta data. 2544 */ 2545 nr_pages = (size - sizeof(struct ring_buffer_cpu_meta)) / 2546 (subbuf_size + sizeof(int)); 2547 /* Need at least two pages plus the reader page */ 2548 if (nr_pages < 3) 2549 goto fail_free_buffers; 2550 2551 again: 2552 /* Make sure that the size fits aligned */ 2553 for (n = 0, ptr = buffers_start; n < nr_cpu_ids; n++) { 2554 ptr += sizeof(struct ring_buffer_cpu_meta) + 2555 sizeof(int) * nr_pages; 2556 ptr = ALIGN(ptr, subbuf_size); 2557 ptr += subbuf_size * nr_pages; 2558 } 2559 if (ptr > end) { 2560 if (nr_pages <= 3) 2561 goto fail_free_buffers; 2562 nr_pages--; 2563 goto again; 2564 } 2565 2566 /* nr_pages should not count the reader page */ 2567 nr_pages--; 2568 buffer->range_addr_start = start; 2569 buffer->range_addr_end = end; 2570 2571 rb_range_meta_init(buffer, nr_pages, scratch_size); 2572 } else { 2573 2574 /* need at least two pages */ 2575 nr_pages = DIV_ROUND_UP(size, buffer->subbuf_size); 2576 if (nr_pages < 2) 2577 nr_pages = 2; 2578 } 2579 2580 cpu = raw_smp_processor_id(); 2581 cpumask_set_cpu(cpu, buffer->cpumask); 2582 buffer->buffers[cpu] = rb_allocate_cpu_buffer(buffer, nr_pages, cpu); 2583 if (!buffer->buffers[cpu]) 2584 goto fail_free_buffers; 2585 2586 ret = cpuhp_state_add_instance(CPUHP_TRACE_RB_PREPARE, &buffer->node); 2587 if (ret < 0) 2588 goto fail_free_buffers; 2589 2590 mutex_init(&buffer->mutex); 2591 2592 return_ptr(buffer); 2593 2594 fail_free_buffers: 2595 for_each_buffer_cpu(buffer, cpu) { 2596 if (buffer->buffers[cpu]) 2597 rb_free_cpu_buffer(buffer->buffers[cpu]); 2598 } 2599 kfree(buffer->buffers); 2600 2601 fail_free_cpumask: 2602 free_cpumask_var(buffer->cpumask); 2603 2604 return NULL; 2605} 2606 2607/** 2608 * __ring_buffer_alloc - allocate a new ring_buffer 2609 * @size: the size in bytes per cpu that is needed. 2610 * @flags: attributes to set for the ring buffer. 2611 * @key: ring buffer reader_lock_key. 2612 * 2613 * Currently the only flag that is available is the RB_FL_OVERWRITE 2614 * flag. This flag means that the buffer will overwrite old data 2615 * when the buffer wraps. If this flag is not set, the buffer will 2616 * drop data when the tail hits the head. 2617 */ 2618struct trace_buffer *__ring_buffer_alloc(unsigned long size, unsigned flags, 2619 struct lock_class_key *key) 2620{ 2621 /* Default buffer page size - one system page */ 2622 return alloc_buffer(size, flags, 0, 0, 0, 0, key); 2623 2624} 2625EXPORT_SYMBOL_GPL(__ring_buffer_alloc); 2626 2627/** 2628 * __ring_buffer_alloc_range - allocate a new ring_buffer from existing memory 2629 * @size: the size in bytes per cpu that is needed. 2630 * @flags: attributes to set for the ring buffer. 2631 * @order: sub-buffer order 2632 * @start: start of allocated range 2633 * @range_size: size of allocated range 2634 * @scratch_size: size of scratch area (for preallocated memory buffers) 2635 * @key: ring buffer reader_lock_key. 2636 * 2637 * Currently the only flag that is available is the RB_FL_OVERWRITE 2638 * flag. This flag means that the buffer will overwrite old data 2639 * when the buffer wraps. If this flag is not set, the buffer will 2640 * drop data when the tail hits the head. 2641 */ 2642struct trace_buffer *__ring_buffer_alloc_range(unsigned long size, unsigned flags, 2643 int order, unsigned long start, 2644 unsigned long range_size, 2645 unsigned long scratch_size, 2646 struct lock_class_key *key) 2647{ 2648 return alloc_buffer(size, flags, order, start, start + range_size, 2649 scratch_size, key); 2650} 2651 2652void *ring_buffer_meta_scratch(struct trace_buffer *buffer, unsigned int *size) 2653{ 2654 struct ring_buffer_meta *meta; 2655 void *ptr; 2656 2657 if (!buffer || !buffer->meta) 2658 return NULL; 2659 2660 meta = buffer->meta; 2661 2662 ptr = (void *)ALIGN((unsigned long)meta + sizeof(*meta), sizeof(long)); 2663 2664 if (size) 2665 *size = (void *)meta + meta->buffers_offset - ptr; 2666 2667 return ptr; 2668} 2669 2670/** 2671 * ring_buffer_free - free a ring buffer. 2672 * @buffer: the buffer to free. 2673 */ 2674void 2675ring_buffer_free(struct trace_buffer *buffer) 2676{ 2677 int cpu; 2678 2679 cpuhp_state_remove_instance(CPUHP_TRACE_RB_PREPARE, &buffer->node); 2680 2681 irq_work_sync(&buffer->irq_work.work); 2682 2683 for_each_buffer_cpu(buffer, cpu) 2684 rb_free_cpu_buffer(buffer->buffers[cpu]); 2685 2686 kfree(buffer->buffers); 2687 free_cpumask_var(buffer->cpumask); 2688 2689 kfree(buffer); 2690} 2691EXPORT_SYMBOL_GPL(ring_buffer_free); 2692 2693void ring_buffer_set_clock(struct trace_buffer *buffer, 2694 u64 (*clock)(void)) 2695{ 2696 buffer->clock = clock; 2697} 2698 2699void ring_buffer_set_time_stamp_abs(struct trace_buffer *buffer, bool abs) 2700{ 2701 buffer->time_stamp_abs = abs; 2702} 2703 2704bool ring_buffer_time_stamp_abs(struct trace_buffer *buffer) 2705{ 2706 return buffer->time_stamp_abs; 2707} 2708 2709static inline unsigned long rb_page_entries(struct buffer_page *bpage) 2710{ 2711 return local_read(&bpage->entries) & RB_WRITE_MASK; 2712} 2713 2714static inline unsigned long rb_page_write(struct buffer_page *bpage) 2715{ 2716 return local_read(&bpage->write) & RB_WRITE_MASK; 2717} 2718 2719static bool 2720rb_remove_pages(struct ring_buffer_per_cpu *cpu_buffer, unsigned long nr_pages) 2721{ 2722 struct list_head *tail_page, *to_remove, *next_page; 2723 struct buffer_page *to_remove_page, *tmp_iter_page; 2724 struct buffer_page *last_page, *first_page; 2725 unsigned long nr_removed; 2726 unsigned long head_bit; 2727 int page_entries; 2728 2729 head_bit = 0; 2730 2731 raw_spin_lock_irq(&cpu_buffer->reader_lock); 2732 atomic_inc(&cpu_buffer->record_disabled); 2733 /* 2734 * We don't race with the readers since we have acquired the reader 2735 * lock. We also don't race with writers after disabling recording. 2736 * This makes it easy to figure out the first and the last page to be 2737 * removed from the list. We unlink all the pages in between including 2738 * the first and last pages. This is done in a busy loop so that we 2739 * lose the least number of traces. 2740 * The pages are freed after we restart recording and unlock readers. 2741 */ 2742 tail_page = &cpu_buffer->tail_page->list; 2743 2744 /* 2745 * tail page might be on reader page, we remove the next page 2746 * from the ring buffer 2747 */ 2748 if (cpu_buffer->tail_page == cpu_buffer->reader_page) 2749 tail_page = rb_list_head(tail_page->next); 2750 to_remove = tail_page; 2751 2752 /* start of pages to remove */ 2753 first_page = list_entry(rb_list_head(to_remove->next), 2754 struct buffer_page, list); 2755 2756 for (nr_removed = 0; nr_removed < nr_pages; nr_removed++) { 2757 to_remove = rb_list_head(to_remove)->next; 2758 head_bit |= (unsigned long)to_remove & RB_PAGE_HEAD; 2759 } 2760 /* Read iterators need to reset themselves when some pages removed */ 2761 cpu_buffer->pages_removed += nr_removed; 2762 2763 next_page = rb_list_head(to_remove)->next; 2764 2765 /* 2766 * Now we remove all pages between tail_page and next_page. 2767 * Make sure that we have head_bit value preserved for the 2768 * next page 2769 */ 2770 tail_page->next = (struct list_head *)((unsigned long)next_page | 2771 head_bit); 2772 next_page = rb_list_head(next_page); 2773 next_page->prev = tail_page; 2774 2775 /* make sure pages points to a valid page in the ring buffer */ 2776 cpu_buffer->pages = next_page; 2777 cpu_buffer->cnt++; 2778 2779 /* update head page */ 2780 if (head_bit) 2781 cpu_buffer->head_page = list_entry(next_page, 2782 struct buffer_page, list); 2783 2784 /* pages are removed, resume tracing and then free the pages */ 2785 atomic_dec(&cpu_buffer->record_disabled); 2786 raw_spin_unlock_irq(&cpu_buffer->reader_lock); 2787 2788 RB_WARN_ON(cpu_buffer, list_empty(cpu_buffer->pages)); 2789 2790 /* last buffer page to remove */ 2791 last_page = list_entry(rb_list_head(to_remove), struct buffer_page, 2792 list); 2793 tmp_iter_page = first_page; 2794 2795 do { 2796 cond_resched(); 2797 2798 to_remove_page = tmp_iter_page; 2799 rb_inc_page(&tmp_iter_page); 2800 2801 /* update the counters */ 2802 page_entries = rb_page_entries(to_remove_page); 2803 if (page_entries) { 2804 /* 2805 * If something was added to this page, it was full 2806 * since it is not the tail page. So we deduct the 2807 * bytes consumed in ring buffer from here. 2808 * Increment overrun to account for the lost events. 2809 */ 2810 local_add(page_entries, &cpu_buffer->overrun); 2811 local_sub(rb_page_commit(to_remove_page), &cpu_buffer->entries_bytes); 2812 local_inc(&cpu_buffer->pages_lost); 2813 } 2814 2815 /* 2816 * We have already removed references to this list item, just 2817 * free up the buffer_page and its page 2818 */ 2819 free_buffer_page(to_remove_page); 2820 nr_removed--; 2821 2822 } while (to_remove_page != last_page); 2823 2824 RB_WARN_ON(cpu_buffer, nr_removed); 2825 2826 return nr_removed == 0; 2827} 2828 2829static bool 2830rb_insert_pages(struct ring_buffer_per_cpu *cpu_buffer) 2831{ 2832 struct list_head *pages = &cpu_buffer->new_pages; 2833 unsigned long flags; 2834 bool success; 2835 int retries; 2836 2837 /* Can be called at early boot up, where interrupts must not been enabled */ 2838 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 2839 /* 2840 * We are holding the reader lock, so the reader page won't be swapped 2841 * in the ring buffer. Now we are racing with the writer trying to 2842 * move head page and the tail page. 2843 * We are going to adapt the reader page update process where: 2844 * 1. We first splice the start and end of list of new pages between 2845 * the head page and its previous page. 2846 * 2. We cmpxchg the prev_page->next to point from head page to the 2847 * start of new pages list. 2848 * 3. Finally, we update the head->prev to the end of new list. 2849 * 2850 * We will try this process 10 times, to make sure that we don't keep 2851 * spinning. 2852 */ 2853 retries = 10; 2854 success = false; 2855 while (retries--) { 2856 struct list_head *head_page, *prev_page; 2857 struct list_head *last_page, *first_page; 2858 struct list_head *head_page_with_bit; 2859 struct buffer_page *hpage = rb_set_head_page(cpu_buffer); 2860 2861 if (!hpage) 2862 break; 2863 head_page = &hpage->list; 2864 prev_page = head_page->prev; 2865 2866 first_page = pages->next; 2867 last_page = pages->prev; 2868 2869 head_page_with_bit = (struct list_head *) 2870 ((unsigned long)head_page | RB_PAGE_HEAD); 2871 2872 last_page->next = head_page_with_bit; 2873 first_page->prev = prev_page; 2874 2875 /* caution: head_page_with_bit gets updated on cmpxchg failure */ 2876 if (try_cmpxchg(&prev_page->next, 2877 &head_page_with_bit, first_page)) { 2878 /* 2879 * yay, we replaced the page pointer to our new list, 2880 * now, we just have to update to head page's prev 2881 * pointer to point to end of list 2882 */ 2883 head_page->prev = last_page; 2884 cpu_buffer->cnt++; 2885 success = true; 2886 break; 2887 } 2888 } 2889 2890 if (success) 2891 INIT_LIST_HEAD(pages); 2892 /* 2893 * If we weren't successful in adding in new pages, warn and stop 2894 * tracing 2895 */ 2896 RB_WARN_ON(cpu_buffer, !success); 2897 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 2898 2899 /* free pages if they weren't inserted */ 2900 if (!success) { 2901 struct buffer_page *bpage, *tmp; 2902 list_for_each_entry_safe(bpage, tmp, &cpu_buffer->new_pages, 2903 list) { 2904 list_del_init(&bpage->list); 2905 free_buffer_page(bpage); 2906 } 2907 } 2908 return success; 2909} 2910 2911static void rb_update_pages(struct ring_buffer_per_cpu *cpu_buffer) 2912{ 2913 bool success; 2914 2915 if (cpu_buffer->nr_pages_to_update > 0) 2916 success = rb_insert_pages(cpu_buffer); 2917 else 2918 success = rb_remove_pages(cpu_buffer, 2919 -cpu_buffer->nr_pages_to_update); 2920 2921 if (success) 2922 cpu_buffer->nr_pages += cpu_buffer->nr_pages_to_update; 2923} 2924 2925static void update_pages_handler(struct work_struct *work) 2926{ 2927 struct ring_buffer_per_cpu *cpu_buffer = container_of(work, 2928 struct ring_buffer_per_cpu, update_pages_work); 2929 rb_update_pages(cpu_buffer); 2930 complete(&cpu_buffer->update_done); 2931} 2932 2933/** 2934 * ring_buffer_resize - resize the ring buffer 2935 * @buffer: the buffer to resize. 2936 * @size: the new size. 2937 * @cpu_id: the cpu buffer to resize 2938 * 2939 * Minimum size is 2 * buffer->subbuf_size. 2940 * 2941 * Returns 0 on success and < 0 on failure. 2942 */ 2943int ring_buffer_resize(struct trace_buffer *buffer, unsigned long size, 2944 int cpu_id) 2945{ 2946 struct ring_buffer_per_cpu *cpu_buffer; 2947 unsigned long nr_pages; 2948 int cpu, err; 2949 2950 /* 2951 * Always succeed at resizing a non-existent buffer: 2952 */ 2953 if (!buffer) 2954 return 0; 2955 2956 /* Make sure the requested buffer exists */ 2957 if (cpu_id != RING_BUFFER_ALL_CPUS && 2958 !cpumask_test_cpu(cpu_id, buffer->cpumask)) 2959 return 0; 2960 2961 nr_pages = DIV_ROUND_UP(size, buffer->subbuf_size); 2962 2963 /* we need a minimum of two pages */ 2964 if (nr_pages < 2) 2965 nr_pages = 2; 2966 2967 /* 2968 * Keep CPUs from coming online while resizing to synchronize 2969 * with new per CPU buffers being created. 2970 */ 2971 guard(cpus_read_lock)(); 2972 2973 /* prevent another thread from changing buffer sizes */ 2974 mutex_lock(&buffer->mutex); 2975 atomic_inc(&buffer->resizing); 2976 2977 if (cpu_id == RING_BUFFER_ALL_CPUS) { 2978 /* 2979 * Don't succeed if resizing is disabled, as a reader might be 2980 * manipulating the ring buffer and is expecting a sane state while 2981 * this is true. 2982 */ 2983 for_each_buffer_cpu(buffer, cpu) { 2984 cpu_buffer = buffer->buffers[cpu]; 2985 if (atomic_read(&cpu_buffer->resize_disabled)) { 2986 err = -EBUSY; 2987 goto out_err_unlock; 2988 } 2989 } 2990 2991 /* calculate the pages to update */ 2992 for_each_buffer_cpu(buffer, cpu) { 2993 cpu_buffer = buffer->buffers[cpu]; 2994 2995 cpu_buffer->nr_pages_to_update = nr_pages - 2996 cpu_buffer->nr_pages; 2997 /* 2998 * nothing more to do for removing pages or no update 2999 */ 3000 if (cpu_buffer->nr_pages_to_update <= 0) 3001 continue; 3002 /* 3003 * to add pages, make sure all new pages can be 3004 * allocated without receiving ENOMEM 3005 */ 3006 INIT_LIST_HEAD(&cpu_buffer->new_pages); 3007 if (__rb_allocate_pages(cpu_buffer, cpu_buffer->nr_pages_to_update, 3008 &cpu_buffer->new_pages)) { 3009 /* not enough memory for new pages */ 3010 err = -ENOMEM; 3011 goto out_err; 3012 } 3013 3014 cond_resched(); 3015 } 3016 3017 /* 3018 * Fire off all the required work handlers 3019 * We can't schedule on offline CPUs, but it's not necessary 3020 * since we can change their buffer sizes without any race. 3021 */ 3022 for_each_buffer_cpu(buffer, cpu) { 3023 cpu_buffer = buffer->buffers[cpu]; 3024 if (!cpu_buffer->nr_pages_to_update) 3025 continue; 3026 3027 /* Can't run something on an offline CPU. */ 3028 if (!cpu_online(cpu)) { 3029 rb_update_pages(cpu_buffer); 3030 cpu_buffer->nr_pages_to_update = 0; 3031 } else { 3032 /* Run directly if possible. */ 3033 migrate_disable(); 3034 if (cpu != smp_processor_id()) { 3035 migrate_enable(); 3036 schedule_work_on(cpu, 3037 &cpu_buffer->update_pages_work); 3038 } else { 3039 update_pages_handler(&cpu_buffer->update_pages_work); 3040 migrate_enable(); 3041 } 3042 } 3043 } 3044 3045 /* wait for all the updates to complete */ 3046 for_each_buffer_cpu(buffer, cpu) { 3047 cpu_buffer = buffer->buffers[cpu]; 3048 if (!cpu_buffer->nr_pages_to_update) 3049 continue; 3050 3051 if (cpu_online(cpu)) 3052 wait_for_completion(&cpu_buffer->update_done); 3053 cpu_buffer->nr_pages_to_update = 0; 3054 } 3055 3056 } else { 3057 cpu_buffer = buffer->buffers[cpu_id]; 3058 3059 if (nr_pages == cpu_buffer->nr_pages) 3060 goto out; 3061 3062 /* 3063 * Don't succeed if resizing is disabled, as a reader might be 3064 * manipulating the ring buffer and is expecting a sane state while 3065 * this is true. 3066 */ 3067 if (atomic_read(&cpu_buffer->resize_disabled)) { 3068 err = -EBUSY; 3069 goto out_err_unlock; 3070 } 3071 3072 cpu_buffer->nr_pages_to_update = nr_pages - 3073 cpu_buffer->nr_pages; 3074 3075 INIT_LIST_HEAD(&cpu_buffer->new_pages); 3076 if (cpu_buffer->nr_pages_to_update > 0 && 3077 __rb_allocate_pages(cpu_buffer, cpu_buffer->nr_pages_to_update, 3078 &cpu_buffer->new_pages)) { 3079 err = -ENOMEM; 3080 goto out_err; 3081 } 3082 3083 /* Can't run something on an offline CPU. */ 3084 if (!cpu_online(cpu_id)) 3085 rb_update_pages(cpu_buffer); 3086 else { 3087 /* Run directly if possible. */ 3088 migrate_disable(); 3089 if (cpu_id == smp_processor_id()) { 3090 rb_update_pages(cpu_buffer); 3091 migrate_enable(); 3092 } else { 3093 migrate_enable(); 3094 schedule_work_on(cpu_id, 3095 &cpu_buffer->update_pages_work); 3096 wait_for_completion(&cpu_buffer->update_done); 3097 } 3098 } 3099 3100 cpu_buffer->nr_pages_to_update = 0; 3101 } 3102 3103 out: 3104 /* 3105 * The ring buffer resize can happen with the ring buffer 3106 * enabled, so that the update disturbs the tracing as little 3107 * as possible. But if the buffer is disabled, we do not need 3108 * to worry about that, and we can take the time to verify 3109 * that the buffer is not corrupt. 3110 */ 3111 if (atomic_read(&buffer->record_disabled)) { 3112 atomic_inc(&buffer->record_disabled); 3113 /* 3114 * Even though the buffer was disabled, we must make sure 3115 * that it is truly disabled before calling rb_check_pages. 3116 * There could have been a race between checking 3117 * record_disable and incrementing it. 3118 */ 3119 synchronize_rcu(); 3120 for_each_buffer_cpu(buffer, cpu) { 3121 cpu_buffer = buffer->buffers[cpu]; 3122 rb_check_pages(cpu_buffer); 3123 } 3124 atomic_dec(&buffer->record_disabled); 3125 } 3126 3127 atomic_dec(&buffer->resizing); 3128 mutex_unlock(&buffer->mutex); 3129 return 0; 3130 3131 out_err: 3132 for_each_buffer_cpu(buffer, cpu) { 3133 struct buffer_page *bpage, *tmp; 3134 3135 cpu_buffer = buffer->buffers[cpu]; 3136 cpu_buffer->nr_pages_to_update = 0; 3137 3138 if (list_empty(&cpu_buffer->new_pages)) 3139 continue; 3140 3141 list_for_each_entry_safe(bpage, tmp, &cpu_buffer->new_pages, 3142 list) { 3143 list_del_init(&bpage->list); 3144 free_buffer_page(bpage); 3145 3146 cond_resched(); 3147 } 3148 } 3149 out_err_unlock: 3150 atomic_dec(&buffer->resizing); 3151 mutex_unlock(&buffer->mutex); 3152 return err; 3153} 3154EXPORT_SYMBOL_GPL(ring_buffer_resize); 3155 3156void ring_buffer_change_overwrite(struct trace_buffer *buffer, int val) 3157{ 3158 mutex_lock(&buffer->mutex); 3159 if (val) 3160 buffer->flags |= RB_FL_OVERWRITE; 3161 else 3162 buffer->flags &= ~RB_FL_OVERWRITE; 3163 mutex_unlock(&buffer->mutex); 3164} 3165EXPORT_SYMBOL_GPL(ring_buffer_change_overwrite); 3166 3167static __always_inline void *__rb_page_index(struct buffer_page *bpage, unsigned index) 3168{ 3169 return bpage->page->data + index; 3170} 3171 3172static __always_inline struct ring_buffer_event * 3173rb_reader_event(struct ring_buffer_per_cpu *cpu_buffer) 3174{ 3175 return __rb_page_index(cpu_buffer->reader_page, 3176 cpu_buffer->reader_page->read); 3177} 3178 3179static struct ring_buffer_event * 3180rb_iter_head_event(struct ring_buffer_iter *iter) 3181{ 3182 struct ring_buffer_event *event; 3183 struct buffer_page *iter_head_page = iter->head_page; 3184 unsigned long commit; 3185 unsigned length; 3186 3187 if (iter->head != iter->next_event) 3188 return iter->event; 3189 3190 /* 3191 * When the writer goes across pages, it issues a cmpxchg which 3192 * is a mb(), which will synchronize with the rmb here. 3193 * (see rb_tail_page_update() and __rb_reserve_next()) 3194 */ 3195 commit = rb_page_commit(iter_head_page); 3196 smp_rmb(); 3197 3198 /* An event needs to be at least 8 bytes in size */ 3199 if (iter->head > commit - 8) 3200 goto reset; 3201 3202 event = __rb_page_index(iter_head_page, iter->head); 3203 length = rb_event_length(event); 3204 3205 /* 3206 * READ_ONCE() doesn't work on functions and we don't want the 3207 * compiler doing any crazy optimizations with length. 3208 */ 3209 barrier(); 3210 3211 if ((iter->head + length) > commit || length > iter->event_size) 3212 /* Writer corrupted the read? */ 3213 goto reset; 3214 3215 memcpy(iter->event, event, length); 3216 /* 3217 * If the page stamp is still the same after this rmb() then the 3218 * event was safely copied without the writer entering the page. 3219 */ 3220 smp_rmb(); 3221 3222 /* Make sure the page didn't change since we read this */ 3223 if (iter->page_stamp != iter_head_page->page->time_stamp || 3224 commit > rb_page_commit(iter_head_page)) 3225 goto reset; 3226 3227 iter->next_event = iter->head + length; 3228 return iter->event; 3229 reset: 3230 /* Reset to the beginning */ 3231 iter->page_stamp = iter->read_stamp = iter->head_page->page->time_stamp; 3232 iter->head = 0; 3233 iter->next_event = 0; 3234 iter->missed_events = 1; 3235 return NULL; 3236} 3237 3238/* Size is determined by what has been committed */ 3239static __always_inline unsigned rb_page_size(struct buffer_page *bpage) 3240{ 3241 return rb_page_commit(bpage) & ~RB_MISSED_MASK; 3242} 3243 3244static __always_inline unsigned 3245rb_commit_index(struct ring_buffer_per_cpu *cpu_buffer) 3246{ 3247 return rb_page_commit(cpu_buffer->commit_page); 3248} 3249 3250static __always_inline unsigned 3251rb_event_index(struct ring_buffer_per_cpu *cpu_buffer, struct ring_buffer_event *event) 3252{ 3253 unsigned long addr = (unsigned long)event; 3254 3255 addr &= (PAGE_SIZE << cpu_buffer->buffer->subbuf_order) - 1; 3256 3257 return addr - BUF_PAGE_HDR_SIZE; 3258} 3259 3260static void rb_inc_iter(struct ring_buffer_iter *iter) 3261{ 3262 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer; 3263 3264 /* 3265 * The iterator could be on the reader page (it starts there). 3266 * But the head could have moved, since the reader was 3267 * found. Check for this case and assign the iterator 3268 * to the head page instead of next. 3269 */ 3270 if (iter->head_page == cpu_buffer->reader_page) 3271 iter->head_page = rb_set_head_page(cpu_buffer); 3272 else 3273 rb_inc_page(&iter->head_page); 3274 3275 iter->page_stamp = iter->read_stamp = iter->head_page->page->time_stamp; 3276 iter->head = 0; 3277 iter->next_event = 0; 3278} 3279 3280/* Return the index into the sub-buffers for a given sub-buffer */ 3281static int rb_meta_subbuf_idx(struct ring_buffer_cpu_meta *meta, void *subbuf) 3282{ 3283 void *subbuf_array; 3284 3285 subbuf_array = (void *)meta + sizeof(int) * meta->nr_subbufs; 3286 subbuf_array = (void *)ALIGN((unsigned long)subbuf_array, meta->subbuf_size); 3287 return (subbuf - subbuf_array) / meta->subbuf_size; 3288} 3289 3290static void rb_update_meta_head(struct ring_buffer_per_cpu *cpu_buffer, 3291 struct buffer_page *next_page) 3292{ 3293 struct ring_buffer_cpu_meta *meta = cpu_buffer->ring_meta; 3294 unsigned long old_head = (unsigned long)next_page->page; 3295 unsigned long new_head; 3296 3297 rb_inc_page(&next_page); 3298 new_head = (unsigned long)next_page->page; 3299 3300 /* 3301 * Only move it forward once, if something else came in and 3302 * moved it forward, then we don't want to touch it. 3303 */ 3304 (void)cmpxchg(&meta->head_buffer, old_head, new_head); 3305} 3306 3307static void rb_update_meta_reader(struct ring_buffer_per_cpu *cpu_buffer, 3308 struct buffer_page *reader) 3309{ 3310 struct ring_buffer_cpu_meta *meta = cpu_buffer->ring_meta; 3311 void *old_reader = cpu_buffer->reader_page->page; 3312 void *new_reader = reader->page; 3313 int id; 3314 3315 id = reader->id; 3316 cpu_buffer->reader_page->id = id; 3317 reader->id = 0; 3318 3319 meta->buffers[0] = rb_meta_subbuf_idx(meta, new_reader); 3320 meta->buffers[id] = rb_meta_subbuf_idx(meta, old_reader); 3321 3322 /* The head pointer is the one after the reader */ 3323 rb_update_meta_head(cpu_buffer, reader); 3324} 3325 3326/* 3327 * rb_handle_head_page - writer hit the head page 3328 * 3329 * Returns: +1 to retry page 3330 * 0 to continue 3331 * -1 on error 3332 */ 3333static int 3334rb_handle_head_page(struct ring_buffer_per_cpu *cpu_buffer, 3335 struct buffer_page *tail_page, 3336 struct buffer_page *next_page) 3337{ 3338 struct buffer_page *new_head; 3339 int entries; 3340 int type; 3341 int ret; 3342 3343 entries = rb_page_entries(next_page); 3344 3345 /* 3346 * The hard part is here. We need to move the head 3347 * forward, and protect against both readers on 3348 * other CPUs and writers coming in via interrupts. 3349 */ 3350 type = rb_head_page_set_update(cpu_buffer, next_page, tail_page, 3351 RB_PAGE_HEAD); 3352 3353 /* 3354 * type can be one of four: 3355 * NORMAL - an interrupt already moved it for us 3356 * HEAD - we are the first to get here. 3357 * UPDATE - we are the interrupt interrupting 3358 * a current move. 3359 * MOVED - a reader on another CPU moved the next 3360 * pointer to its reader page. Give up 3361 * and try again. 3362 */ 3363 3364 switch (type) { 3365 case RB_PAGE_HEAD: 3366 /* 3367 * We changed the head to UPDATE, thus 3368 * it is our responsibility to update 3369 * the counters. 3370 */ 3371 local_add(entries, &cpu_buffer->overrun); 3372 local_sub(rb_page_commit(next_page), &cpu_buffer->entries_bytes); 3373 local_inc(&cpu_buffer->pages_lost); 3374 3375 if (cpu_buffer->ring_meta) 3376 rb_update_meta_head(cpu_buffer, next_page); 3377 /* 3378 * The entries will be zeroed out when we move the 3379 * tail page. 3380 */ 3381 3382 /* still more to do */ 3383 break; 3384 3385 case RB_PAGE_UPDATE: 3386 /* 3387 * This is an interrupt that interrupt the 3388 * previous update. Still more to do. 3389 */ 3390 break; 3391 case RB_PAGE_NORMAL: 3392 /* 3393 * An interrupt came in before the update 3394 * and processed this for us. 3395 * Nothing left to do. 3396 */ 3397 return 1; 3398 case RB_PAGE_MOVED: 3399 /* 3400 * The reader is on another CPU and just did 3401 * a swap with our next_page. 3402 * Try again. 3403 */ 3404 return 1; 3405 default: 3406 RB_WARN_ON(cpu_buffer, 1); /* WTF??? */ 3407 return -1; 3408 } 3409 3410 /* 3411 * Now that we are here, the old head pointer is 3412 * set to UPDATE. This will keep the reader from 3413 * swapping the head page with the reader page. 3414 * The reader (on another CPU) will spin till 3415 * we are finished. 3416 * 3417 * We just need to protect against interrupts 3418 * doing the job. We will set the next pointer 3419 * to HEAD. After that, we set the old pointer 3420 * to NORMAL, but only if it was HEAD before. 3421 * otherwise we are an interrupt, and only 3422 * want the outer most commit to reset it. 3423 */ 3424 new_head = next_page; 3425 rb_inc_page(&new_head); 3426 3427 ret = rb_head_page_set_head(cpu_buffer, new_head, next_page, 3428 RB_PAGE_NORMAL); 3429 3430 /* 3431 * Valid returns are: 3432 * HEAD - an interrupt came in and already set it. 3433 * NORMAL - One of two things: 3434 * 1) We really set it. 3435 * 2) A bunch of interrupts came in and moved 3436 * the page forward again. 3437 */ 3438 switch (ret) { 3439 case RB_PAGE_HEAD: 3440 case RB_PAGE_NORMAL: 3441 /* OK */ 3442 break; 3443 default: 3444 RB_WARN_ON(cpu_buffer, 1); 3445 return -1; 3446 } 3447 3448 /* 3449 * It is possible that an interrupt came in, 3450 * set the head up, then more interrupts came in 3451 * and moved it again. When we get back here, 3452 * the page would have been set to NORMAL but we 3453 * just set it back to HEAD. 3454 * 3455 * How do you detect this? Well, if that happened 3456 * the tail page would have moved. 3457 */ 3458 if (ret == RB_PAGE_NORMAL) { 3459 struct buffer_page *buffer_tail_page; 3460 3461 buffer_tail_page = READ_ONCE(cpu_buffer->tail_page); 3462 /* 3463 * If the tail had moved passed next, then we need 3464 * to reset the pointer. 3465 */ 3466 if (buffer_tail_page != tail_page && 3467 buffer_tail_page != next_page) 3468 rb_head_page_set_normal(cpu_buffer, new_head, 3469 next_page, 3470 RB_PAGE_HEAD); 3471 } 3472 3473 /* 3474 * If this was the outer most commit (the one that 3475 * changed the original pointer from HEAD to UPDATE), 3476 * then it is up to us to reset it to NORMAL. 3477 */ 3478 if (type == RB_PAGE_HEAD) { 3479 ret = rb_head_page_set_normal(cpu_buffer, next_page, 3480 tail_page, 3481 RB_PAGE_UPDATE); 3482 if (RB_WARN_ON(cpu_buffer, 3483 ret != RB_PAGE_UPDATE)) 3484 return -1; 3485 } 3486 3487 return 0; 3488} 3489 3490static inline void 3491rb_reset_tail(struct ring_buffer_per_cpu *cpu_buffer, 3492 unsigned long tail, struct rb_event_info *info) 3493{ 3494 unsigned long bsize = READ_ONCE(cpu_buffer->buffer->subbuf_size); 3495 struct buffer_page *tail_page = info->tail_page; 3496 struct ring_buffer_event *event; 3497 unsigned long length = info->length; 3498 3499 /* 3500 * Only the event that crossed the page boundary 3501 * must fill the old tail_page with padding. 3502 */ 3503 if (tail >= bsize) { 3504 /* 3505 * If the page was filled, then we still need 3506 * to update the real_end. Reset it to zero 3507 * and the reader will ignore it. 3508 */ 3509 if (tail == bsize) 3510 tail_page->real_end = 0; 3511 3512 local_sub(length, &tail_page->write); 3513 return; 3514 } 3515 3516 event = __rb_page_index(tail_page, tail); 3517 3518 /* 3519 * Save the original length to the meta data. 3520 * This will be used by the reader to add lost event 3521 * counter. 3522 */ 3523 tail_page->real_end = tail; 3524 3525 /* 3526 * If this event is bigger than the minimum size, then 3527 * we need to be careful that we don't subtract the 3528 * write counter enough to allow another writer to slip 3529 * in on this page. 3530 * We put in a discarded commit instead, to make sure 3531 * that this space is not used again, and this space will 3532 * not be accounted into 'entries_bytes'. 3533 * 3534 * If we are less than the minimum size, we don't need to 3535 * worry about it. 3536 */ 3537 if (tail > (bsize - RB_EVNT_MIN_SIZE)) { 3538 /* No room for any events */ 3539 3540 /* Mark the rest of the page with padding */ 3541 rb_event_set_padding(event); 3542 3543 /* Make sure the padding is visible before the write update */ 3544 smp_wmb(); 3545 3546 /* Set the write back to the previous setting */ 3547 local_sub(length, &tail_page->write); 3548 return; 3549 } 3550 3551 /* Put in a discarded event */ 3552 event->array[0] = (bsize - tail) - RB_EVNT_HDR_SIZE; 3553 event->type_len = RINGBUF_TYPE_PADDING; 3554 /* time delta must be non zero */ 3555 event->time_delta = 1; 3556 3557 /* account for padding bytes */ 3558 local_add(bsize - tail, &cpu_buffer->entries_bytes); 3559 3560 /* Make sure the padding is visible before the tail_page->write update */ 3561 smp_wmb(); 3562 3563 /* Set write to end of buffer */ 3564 length = (tail + length) - bsize; 3565 local_sub(length, &tail_page->write); 3566} 3567 3568static inline void rb_end_commit(struct ring_buffer_per_cpu *cpu_buffer); 3569 3570/* 3571 * This is the slow path, force gcc not to inline it. 3572 */ 3573static noinline struct ring_buffer_event * 3574rb_move_tail(struct ring_buffer_per_cpu *cpu_buffer, 3575 unsigned long tail, struct rb_event_info *info) 3576{ 3577 struct buffer_page *tail_page = info->tail_page; 3578 struct buffer_page *commit_page = cpu_buffer->commit_page; 3579 struct trace_buffer *buffer = cpu_buffer->buffer; 3580 struct buffer_page *next_page; 3581 int ret; 3582 3583 next_page = tail_page; 3584 3585 rb_inc_page(&next_page); 3586 3587 /* 3588 * If for some reason, we had an interrupt storm that made 3589 * it all the way around the buffer, bail, and warn 3590 * about it. 3591 */ 3592 if (unlikely(next_page == commit_page)) { 3593 local_inc(&cpu_buffer->commit_overrun); 3594 goto out_reset; 3595 } 3596 3597 /* 3598 * This is where the fun begins! 3599 * 3600 * We are fighting against races between a reader that 3601 * could be on another CPU trying to swap its reader 3602 * page with the buffer head. 3603 * 3604 * We are also fighting against interrupts coming in and 3605 * moving the head or tail on us as well. 3606 * 3607 * If the next page is the head page then we have filled 3608 * the buffer, unless the commit page is still on the 3609 * reader page. 3610 */ 3611 if (rb_is_head_page(next_page, &tail_page->list)) { 3612 3613 /* 3614 * If the commit is not on the reader page, then 3615 * move the header page. 3616 */ 3617 if (!rb_is_reader_page(cpu_buffer->commit_page)) { 3618 /* 3619 * If we are not in overwrite mode, 3620 * this is easy, just stop here. 3621 */ 3622 if (!(buffer->flags & RB_FL_OVERWRITE)) { 3623 local_inc(&cpu_buffer->dropped_events); 3624 goto out_reset; 3625 } 3626 3627 ret = rb_handle_head_page(cpu_buffer, 3628 tail_page, 3629 next_page); 3630 if (ret < 0) 3631 goto out_reset; 3632 if (ret) 3633 goto out_again; 3634 } else { 3635 /* 3636 * We need to be careful here too. The 3637 * commit page could still be on the reader 3638 * page. We could have a small buffer, and 3639 * have filled up the buffer with events 3640 * from interrupts and such, and wrapped. 3641 * 3642 * Note, if the tail page is also on the 3643 * reader_page, we let it move out. 3644 */ 3645 if (unlikely((cpu_buffer->commit_page != 3646 cpu_buffer->tail_page) && 3647 (cpu_buffer->commit_page == 3648 cpu_buffer->reader_page))) { 3649 local_inc(&cpu_buffer->commit_overrun); 3650 goto out_reset; 3651 } 3652 } 3653 } 3654 3655 rb_tail_page_update(cpu_buffer, tail_page, next_page); 3656 3657 out_again: 3658 3659 rb_reset_tail(cpu_buffer, tail, info); 3660 3661 /* Commit what we have for now. */ 3662 rb_end_commit(cpu_buffer); 3663 /* rb_end_commit() decs committing */ 3664 local_inc(&cpu_buffer->committing); 3665 3666 /* fail and let the caller try again */ 3667 return ERR_PTR(-EAGAIN); 3668 3669 out_reset: 3670 /* reset write */ 3671 rb_reset_tail(cpu_buffer, tail, info); 3672 3673 return NULL; 3674} 3675 3676/* Slow path */ 3677static struct ring_buffer_event * 3678rb_add_time_stamp(struct ring_buffer_per_cpu *cpu_buffer, 3679 struct ring_buffer_event *event, u64 delta, bool abs) 3680{ 3681 if (abs) 3682 event->type_len = RINGBUF_TYPE_TIME_STAMP; 3683 else 3684 event->type_len = RINGBUF_TYPE_TIME_EXTEND; 3685 3686 /* Not the first event on the page, or not delta? */ 3687 if (abs || rb_event_index(cpu_buffer, event)) { 3688 event->time_delta = delta & TS_MASK; 3689 event->array[0] = delta >> TS_SHIFT; 3690 } else { 3691 /* nope, just zero it */ 3692 event->time_delta = 0; 3693 event->array[0] = 0; 3694 } 3695 3696 return skip_time_extend(event); 3697} 3698 3699#ifndef CONFIG_HAVE_UNSTABLE_SCHED_CLOCK 3700static inline bool sched_clock_stable(void) 3701{ 3702 return true; 3703} 3704#endif 3705 3706static void 3707rb_check_timestamp(struct ring_buffer_per_cpu *cpu_buffer, 3708 struct rb_event_info *info) 3709{ 3710 u64 write_stamp; 3711 3712 WARN_ONCE(1, "Delta way too big! %llu ts=%llu before=%llu after=%llu write stamp=%llu\n%s", 3713 (unsigned long long)info->delta, 3714 (unsigned long long)info->ts, 3715 (unsigned long long)info->before, 3716 (unsigned long long)info->after, 3717 (unsigned long long)({rb_time_read(&cpu_buffer->write_stamp, &write_stamp); write_stamp;}), 3718 sched_clock_stable() ? "" : 3719 "If you just came from a suspend/resume,\n" 3720 "please switch to the trace global clock:\n" 3721 " echo global > /sys/kernel/tracing/trace_clock\n" 3722 "or add trace_clock=global to the kernel command line\n"); 3723} 3724 3725static void rb_add_timestamp(struct ring_buffer_per_cpu *cpu_buffer, 3726 struct ring_buffer_event **event, 3727 struct rb_event_info *info, 3728 u64 *delta, 3729 unsigned int *length) 3730{ 3731 bool abs = info->add_timestamp & 3732 (RB_ADD_STAMP_FORCE | RB_ADD_STAMP_ABSOLUTE); 3733 3734 if (unlikely(info->delta > (1ULL << 59))) { 3735 /* 3736 * Some timers can use more than 59 bits, and when a timestamp 3737 * is added to the buffer, it will lose those bits. 3738 */ 3739 if (abs && (info->ts & TS_MSB)) { 3740 info->delta &= ABS_TS_MASK; 3741 3742 /* did the clock go backwards */ 3743 } else if (info->before == info->after && info->before > info->ts) { 3744 /* not interrupted */ 3745 static int once; 3746 3747 /* 3748 * This is possible with a recalibrating of the TSC. 3749 * Do not produce a call stack, but just report it. 3750 */ 3751 if (!once) { 3752 once++; 3753 pr_warn("Ring buffer clock went backwards: %llu -> %llu\n", 3754 info->before, info->ts); 3755 } 3756 } else 3757 rb_check_timestamp(cpu_buffer, info); 3758 if (!abs) 3759 info->delta = 0; 3760 } 3761 *event = rb_add_time_stamp(cpu_buffer, *event, info->delta, abs); 3762 *length -= RB_LEN_TIME_EXTEND; 3763 *delta = 0; 3764} 3765 3766/** 3767 * rb_update_event - update event type and data 3768 * @cpu_buffer: The per cpu buffer of the @event 3769 * @event: the event to update 3770 * @info: The info to update the @event with (contains length and delta) 3771 * 3772 * Update the type and data fields of the @event. The length 3773 * is the actual size that is written to the ring buffer, 3774 * and with this, we can determine what to place into the 3775 * data field. 3776 */ 3777static void 3778rb_update_event(struct ring_buffer_per_cpu *cpu_buffer, 3779 struct ring_buffer_event *event, 3780 struct rb_event_info *info) 3781{ 3782 unsigned length = info->length; 3783 u64 delta = info->delta; 3784 unsigned int nest = local_read(&cpu_buffer->committing) - 1; 3785 3786 if (!WARN_ON_ONCE(nest >= MAX_NEST)) 3787 cpu_buffer->event_stamp[nest] = info->ts; 3788 3789 /* 3790 * If we need to add a timestamp, then we 3791 * add it to the start of the reserved space. 3792 */ 3793 if (unlikely(info->add_timestamp)) 3794 rb_add_timestamp(cpu_buffer, &event, info, &delta, &length); 3795 3796 event->time_delta = delta; 3797 length -= RB_EVNT_HDR_SIZE; 3798 if (length > RB_MAX_SMALL_DATA || RB_FORCE_8BYTE_ALIGNMENT) { 3799 event->type_len = 0; 3800 event->array[0] = length; 3801 } else 3802 event->type_len = DIV_ROUND_UP(length, RB_ALIGNMENT); 3803} 3804 3805static unsigned rb_calculate_event_length(unsigned length) 3806{ 3807 struct ring_buffer_event event; /* Used only for sizeof array */ 3808 3809 /* zero length can cause confusions */ 3810 if (!length) 3811 length++; 3812 3813 if (length > RB_MAX_SMALL_DATA || RB_FORCE_8BYTE_ALIGNMENT) 3814 length += sizeof(event.array[0]); 3815 3816 length += RB_EVNT_HDR_SIZE; 3817 length = ALIGN(length, RB_ARCH_ALIGNMENT); 3818 3819 /* 3820 * In case the time delta is larger than the 27 bits for it 3821 * in the header, we need to add a timestamp. If another 3822 * event comes in when trying to discard this one to increase 3823 * the length, then the timestamp will be added in the allocated 3824 * space of this event. If length is bigger than the size needed 3825 * for the TIME_EXTEND, then padding has to be used. The events 3826 * length must be either RB_LEN_TIME_EXTEND, or greater than or equal 3827 * to RB_LEN_TIME_EXTEND + 8, as 8 is the minimum size for padding. 3828 * As length is a multiple of 4, we only need to worry if it 3829 * is 12 (RB_LEN_TIME_EXTEND + 4). 3830 */ 3831 if (length == RB_LEN_TIME_EXTEND + RB_ALIGNMENT) 3832 length += RB_ALIGNMENT; 3833 3834 return length; 3835} 3836 3837static inline bool 3838rb_try_to_discard(struct ring_buffer_per_cpu *cpu_buffer, 3839 struct ring_buffer_event *event) 3840{ 3841 unsigned long new_index, old_index; 3842 struct buffer_page *bpage; 3843 unsigned long addr; 3844 3845 new_index = rb_event_index(cpu_buffer, event); 3846 old_index = new_index + rb_event_ts_length(event); 3847 addr = (unsigned long)event; 3848 addr &= ~((PAGE_SIZE << cpu_buffer->buffer->subbuf_order) - 1); 3849 3850 bpage = READ_ONCE(cpu_buffer->tail_page); 3851 3852 /* 3853 * Make sure the tail_page is still the same and 3854 * the next write location is the end of this event 3855 */ 3856 if (bpage->page == (void *)addr && rb_page_write(bpage) == old_index) { 3857 unsigned long write_mask = 3858 local_read(&bpage->write) & ~RB_WRITE_MASK; 3859 unsigned long event_length = rb_event_length(event); 3860 3861 /* 3862 * For the before_stamp to be different than the write_stamp 3863 * to make sure that the next event adds an absolute 3864 * value and does not rely on the saved write stamp, which 3865 * is now going to be bogus. 3866 * 3867 * By setting the before_stamp to zero, the next event 3868 * is not going to use the write_stamp and will instead 3869 * create an absolute timestamp. This means there's no 3870 * reason to update the wirte_stamp! 3871 */ 3872 rb_time_set(&cpu_buffer->before_stamp, 0); 3873 3874 /* 3875 * If an event were to come in now, it would see that the 3876 * write_stamp and the before_stamp are different, and assume 3877 * that this event just added itself before updating 3878 * the write stamp. The interrupting event will fix the 3879 * write stamp for us, and use an absolute timestamp. 3880 */ 3881 3882 /* 3883 * This is on the tail page. It is possible that 3884 * a write could come in and move the tail page 3885 * and write to the next page. That is fine 3886 * because we just shorten what is on this page. 3887 */ 3888 old_index += write_mask; 3889 new_index += write_mask; 3890 3891 /* caution: old_index gets updated on cmpxchg failure */ 3892 if (local_try_cmpxchg(&bpage->write, &old_index, new_index)) { 3893 /* update counters */ 3894 local_sub(event_length, &cpu_buffer->entries_bytes); 3895 return true; 3896 } 3897 } 3898 3899 /* could not discard */ 3900 return false; 3901} 3902 3903static void rb_start_commit(struct ring_buffer_per_cpu *cpu_buffer) 3904{ 3905 local_inc(&cpu_buffer->committing); 3906 local_inc(&cpu_buffer->commits); 3907} 3908 3909static __always_inline void 3910rb_set_commit_to_write(struct ring_buffer_per_cpu *cpu_buffer) 3911{ 3912 unsigned long max_count; 3913 3914 /* 3915 * We only race with interrupts and NMIs on this CPU. 3916 * If we own the commit event, then we can commit 3917 * all others that interrupted us, since the interruptions 3918 * are in stack format (they finish before they come 3919 * back to us). This allows us to do a simple loop to 3920 * assign the commit to the tail. 3921 */ 3922 again: 3923 max_count = cpu_buffer->nr_pages * 100; 3924 3925 while (cpu_buffer->commit_page != READ_ONCE(cpu_buffer->tail_page)) { 3926 if (RB_WARN_ON(cpu_buffer, !(--max_count))) 3927 return; 3928 if (RB_WARN_ON(cpu_buffer, 3929 rb_is_reader_page(cpu_buffer->tail_page))) 3930 return; 3931 /* 3932 * No need for a memory barrier here, as the update 3933 * of the tail_page did it for this page. 3934 */ 3935 local_set(&cpu_buffer->commit_page->page->commit, 3936 rb_page_write(cpu_buffer->commit_page)); 3937 rb_inc_page(&cpu_buffer->commit_page); 3938 if (cpu_buffer->ring_meta) { 3939 struct ring_buffer_cpu_meta *meta = cpu_buffer->ring_meta; 3940 meta->commit_buffer = (unsigned long)cpu_buffer->commit_page->page; 3941 } 3942 /* add barrier to keep gcc from optimizing too much */ 3943 barrier(); 3944 } 3945 while (rb_commit_index(cpu_buffer) != 3946 rb_page_write(cpu_buffer->commit_page)) { 3947 3948 /* Make sure the readers see the content of what is committed. */ 3949 smp_wmb(); 3950 local_set(&cpu_buffer->commit_page->page->commit, 3951 rb_page_write(cpu_buffer->commit_page)); 3952 RB_WARN_ON(cpu_buffer, 3953 local_read(&cpu_buffer->commit_page->page->commit) & 3954 ~RB_WRITE_MASK); 3955 barrier(); 3956 } 3957 3958 /* again, keep gcc from optimizing */ 3959 barrier(); 3960 3961 /* 3962 * If an interrupt came in just after the first while loop 3963 * and pushed the tail page forward, we will be left with 3964 * a dangling commit that will never go forward. 3965 */ 3966 if (unlikely(cpu_buffer->commit_page != READ_ONCE(cpu_buffer->tail_page))) 3967 goto again; 3968} 3969 3970static __always_inline void rb_end_commit(struct ring_buffer_per_cpu *cpu_buffer) 3971{ 3972 unsigned long commits; 3973 3974 if (RB_WARN_ON(cpu_buffer, 3975 !local_read(&cpu_buffer->committing))) 3976 return; 3977 3978 again: 3979 commits = local_read(&cpu_buffer->commits); 3980 /* synchronize with interrupts */ 3981 barrier(); 3982 if (local_read(&cpu_buffer->committing) == 1) 3983 rb_set_commit_to_write(cpu_buffer); 3984 3985 local_dec(&cpu_buffer->committing); 3986 3987 /* synchronize with interrupts */ 3988 barrier(); 3989 3990 /* 3991 * Need to account for interrupts coming in between the 3992 * updating of the commit page and the clearing of the 3993 * committing counter. 3994 */ 3995 if (unlikely(local_read(&cpu_buffer->commits) != commits) && 3996 !local_read(&cpu_buffer->committing)) { 3997 local_inc(&cpu_buffer->committing); 3998 goto again; 3999 } 4000} 4001 4002static inline void rb_event_discard(struct ring_buffer_event *event) 4003{ 4004 if (extended_time(event)) 4005 event = skip_time_extend(event); 4006 4007 /* array[0] holds the actual length for the discarded event */ 4008 event->array[0] = rb_event_data_length(event) - RB_EVNT_HDR_SIZE; 4009 event->type_len = RINGBUF_TYPE_PADDING; 4010 /* time delta must be non zero */ 4011 if (!event->time_delta) 4012 event->time_delta = 1; 4013} 4014 4015static void rb_commit(struct ring_buffer_per_cpu *cpu_buffer) 4016{ 4017 local_inc(&cpu_buffer->entries); 4018 rb_end_commit(cpu_buffer); 4019} 4020 4021static __always_inline void 4022rb_wakeups(struct trace_buffer *buffer, struct ring_buffer_per_cpu *cpu_buffer) 4023{ 4024 if (buffer->irq_work.waiters_pending) { 4025 buffer->irq_work.waiters_pending = false; 4026 /* irq_work_queue() supplies it's own memory barriers */ 4027 irq_work_queue(&buffer->irq_work.work); 4028 } 4029 4030 if (cpu_buffer->irq_work.waiters_pending) { 4031 cpu_buffer->irq_work.waiters_pending = false; 4032 /* irq_work_queue() supplies it's own memory barriers */ 4033 irq_work_queue(&cpu_buffer->irq_work.work); 4034 } 4035 4036 if (cpu_buffer->last_pages_touch == local_read(&cpu_buffer->pages_touched)) 4037 return; 4038 4039 if (cpu_buffer->reader_page == cpu_buffer->commit_page) 4040 return; 4041 4042 if (!cpu_buffer->irq_work.full_waiters_pending) 4043 return; 4044 4045 cpu_buffer->last_pages_touch = local_read(&cpu_buffer->pages_touched); 4046 4047 if (!full_hit(buffer, cpu_buffer->cpu, cpu_buffer->shortest_full)) 4048 return; 4049 4050 cpu_buffer->irq_work.wakeup_full = true; 4051 cpu_buffer->irq_work.full_waiters_pending = false; 4052 /* irq_work_queue() supplies it's own memory barriers */ 4053 irq_work_queue(&cpu_buffer->irq_work.work); 4054} 4055 4056#ifdef CONFIG_RING_BUFFER_RECORD_RECURSION 4057# define do_ring_buffer_record_recursion() \ 4058 do_ftrace_record_recursion(_THIS_IP_, _RET_IP_) 4059#else 4060# define do_ring_buffer_record_recursion() do { } while (0) 4061#endif 4062 4063/* 4064 * The lock and unlock are done within a preempt disable section. 4065 * The current_context per_cpu variable can only be modified 4066 * by the current task between lock and unlock. But it can 4067 * be modified more than once via an interrupt. To pass this 4068 * information from the lock to the unlock without having to 4069 * access the 'in_interrupt()' functions again (which do show 4070 * a bit of overhead in something as critical as function tracing, 4071 * we use a bitmask trick. 4072 * 4073 * bit 1 = NMI context 4074 * bit 2 = IRQ context 4075 * bit 3 = SoftIRQ context 4076 * bit 4 = normal context. 4077 * 4078 * This works because this is the order of contexts that can 4079 * preempt other contexts. A SoftIRQ never preempts an IRQ 4080 * context. 4081 * 4082 * When the context is determined, the corresponding bit is 4083 * checked and set (if it was set, then a recursion of that context 4084 * happened). 4085 * 4086 * On unlock, we need to clear this bit. To do so, just subtract 4087 * 1 from the current_context and AND it to itself. 4088 * 4089 * (binary) 4090 * 101 - 1 = 100 4091 * 101 & 100 = 100 (clearing bit zero) 4092 * 4093 * 1010 - 1 = 1001 4094 * 1010 & 1001 = 1000 (clearing bit 1) 4095 * 4096 * The least significant bit can be cleared this way, and it 4097 * just so happens that it is the same bit corresponding to 4098 * the current context. 4099 * 4100 * Now the TRANSITION bit breaks the above slightly. The TRANSITION bit 4101 * is set when a recursion is detected at the current context, and if 4102 * the TRANSITION bit is already set, it will fail the recursion. 4103 * This is needed because there's a lag between the changing of 4104 * interrupt context and updating the preempt count. In this case, 4105 * a false positive will be found. To handle this, one extra recursion 4106 * is allowed, and this is done by the TRANSITION bit. If the TRANSITION 4107 * bit is already set, then it is considered a recursion and the function 4108 * ends. Otherwise, the TRANSITION bit is set, and that bit is returned. 4109 * 4110 * On the trace_recursive_unlock(), the TRANSITION bit will be the first 4111 * to be cleared. Even if it wasn't the context that set it. That is, 4112 * if an interrupt comes in while NORMAL bit is set and the ring buffer 4113 * is called before preempt_count() is updated, since the check will 4114 * be on the NORMAL bit, the TRANSITION bit will then be set. If an 4115 * NMI then comes in, it will set the NMI bit, but when the NMI code 4116 * does the trace_recursive_unlock() it will clear the TRANSITION bit 4117 * and leave the NMI bit set. But this is fine, because the interrupt 4118 * code that set the TRANSITION bit will then clear the NMI bit when it 4119 * calls trace_recursive_unlock(). If another NMI comes in, it will 4120 * set the TRANSITION bit and continue. 4121 * 4122 * Note: The TRANSITION bit only handles a single transition between context. 4123 */ 4124 4125static __always_inline bool 4126trace_recursive_lock(struct ring_buffer_per_cpu *cpu_buffer) 4127{ 4128 unsigned int val = cpu_buffer->current_context; 4129 int bit = interrupt_context_level(); 4130 4131 bit = RB_CTX_NORMAL - bit; 4132 4133 if (unlikely(val & (1 << (bit + cpu_buffer->nest)))) { 4134 /* 4135 * It is possible that this was called by transitioning 4136 * between interrupt context, and preempt_count() has not 4137 * been updated yet. In this case, use the TRANSITION bit. 4138 */ 4139 bit = RB_CTX_TRANSITION; 4140 if (val & (1 << (bit + cpu_buffer->nest))) { 4141 do_ring_buffer_record_recursion(); 4142 return true; 4143 } 4144 } 4145 4146 val |= (1 << (bit + cpu_buffer->nest)); 4147 cpu_buffer->current_context = val; 4148 4149 return false; 4150} 4151 4152static __always_inline void 4153trace_recursive_unlock(struct ring_buffer_per_cpu *cpu_buffer) 4154{ 4155 cpu_buffer->current_context &= 4156 cpu_buffer->current_context - (1 << cpu_buffer->nest); 4157} 4158 4159/* The recursive locking above uses 5 bits */ 4160#define NESTED_BITS 5 4161 4162/** 4163 * ring_buffer_nest_start - Allow to trace while nested 4164 * @buffer: The ring buffer to modify 4165 * 4166 * The ring buffer has a safety mechanism to prevent recursion. 4167 * But there may be a case where a trace needs to be done while 4168 * tracing something else. In this case, calling this function 4169 * will allow this function to nest within a currently active 4170 * ring_buffer_lock_reserve(). 4171 * 4172 * Call this function before calling another ring_buffer_lock_reserve() and 4173 * call ring_buffer_nest_end() after the nested ring_buffer_unlock_commit(). 4174 */ 4175void ring_buffer_nest_start(struct trace_buffer *buffer) 4176{ 4177 struct ring_buffer_per_cpu *cpu_buffer; 4178 int cpu; 4179 4180 /* Enabled by ring_buffer_nest_end() */ 4181 preempt_disable_notrace(); 4182 cpu = raw_smp_processor_id(); 4183 cpu_buffer = buffer->buffers[cpu]; 4184 /* This is the shift value for the above recursive locking */ 4185 cpu_buffer->nest += NESTED_BITS; 4186} 4187 4188/** 4189 * ring_buffer_nest_end - Allow to trace while nested 4190 * @buffer: The ring buffer to modify 4191 * 4192 * Must be called after ring_buffer_nest_start() and after the 4193 * ring_buffer_unlock_commit(). 4194 */ 4195void ring_buffer_nest_end(struct trace_buffer *buffer) 4196{ 4197 struct ring_buffer_per_cpu *cpu_buffer; 4198 int cpu; 4199 4200 /* disabled by ring_buffer_nest_start() */ 4201 cpu = raw_smp_processor_id(); 4202 cpu_buffer = buffer->buffers[cpu]; 4203 /* This is the shift value for the above recursive locking */ 4204 cpu_buffer->nest -= NESTED_BITS; 4205 preempt_enable_notrace(); 4206} 4207 4208/** 4209 * ring_buffer_unlock_commit - commit a reserved 4210 * @buffer: The buffer to commit to 4211 * 4212 * This commits the data to the ring buffer, and releases any locks held. 4213 * 4214 * Must be paired with ring_buffer_lock_reserve. 4215 */ 4216int ring_buffer_unlock_commit(struct trace_buffer *buffer) 4217{ 4218 struct ring_buffer_per_cpu *cpu_buffer; 4219 int cpu = raw_smp_processor_id(); 4220 4221 cpu_buffer = buffer->buffers[cpu]; 4222 4223 rb_commit(cpu_buffer); 4224 4225 rb_wakeups(buffer, cpu_buffer); 4226 4227 trace_recursive_unlock(cpu_buffer); 4228 4229 preempt_enable_notrace(); 4230 4231 return 0; 4232} 4233EXPORT_SYMBOL_GPL(ring_buffer_unlock_commit); 4234 4235/* Special value to validate all deltas on a page. */ 4236#define CHECK_FULL_PAGE 1L 4237 4238#ifdef CONFIG_RING_BUFFER_VALIDATE_TIME_DELTAS 4239 4240static const char *show_irq_str(int bits) 4241{ 4242 static const char * type[] = { 4243 ".", // 0 4244 "s", // 1 4245 "h", // 2 4246 "Hs", // 3 4247 "n", // 4 4248 "Ns", // 5 4249 "Nh", // 6 4250 "NHs", // 7 4251 }; 4252 4253 return type[bits]; 4254} 4255 4256/* Assume this is a trace event */ 4257static const char *show_flags(struct ring_buffer_event *event) 4258{ 4259 struct trace_entry *entry; 4260 int bits = 0; 4261 4262 if (rb_event_data_length(event) - RB_EVNT_HDR_SIZE < sizeof(*entry)) 4263 return "X"; 4264 4265 entry = ring_buffer_event_data(event); 4266 4267 if (entry->flags & TRACE_FLAG_SOFTIRQ) 4268 bits |= 1; 4269 4270 if (entry->flags & TRACE_FLAG_HARDIRQ) 4271 bits |= 2; 4272 4273 if (entry->flags & TRACE_FLAG_NMI) 4274 bits |= 4; 4275 4276 return show_irq_str(bits); 4277} 4278 4279static const char *show_irq(struct ring_buffer_event *event) 4280{ 4281 struct trace_entry *entry; 4282 4283 if (rb_event_data_length(event) - RB_EVNT_HDR_SIZE < sizeof(*entry)) 4284 return ""; 4285 4286 entry = ring_buffer_event_data(event); 4287 if (entry->flags & TRACE_FLAG_IRQS_OFF) 4288 return "d"; 4289 return ""; 4290} 4291 4292static const char *show_interrupt_level(void) 4293{ 4294 unsigned long pc = preempt_count(); 4295 unsigned char level = 0; 4296 4297 if (pc & SOFTIRQ_OFFSET) 4298 level |= 1; 4299 4300 if (pc & HARDIRQ_MASK) 4301 level |= 2; 4302 4303 if (pc & NMI_MASK) 4304 level |= 4; 4305 4306 return show_irq_str(level); 4307} 4308 4309static void dump_buffer_page(struct buffer_data_page *bpage, 4310 struct rb_event_info *info, 4311 unsigned long tail) 4312{ 4313 struct ring_buffer_event *event; 4314 u64 ts, delta; 4315 int e; 4316 4317 ts = bpage->time_stamp; 4318 pr_warn(" [%lld] PAGE TIME STAMP\n", ts); 4319 4320 for (e = 0; e < tail; e += rb_event_length(event)) { 4321 4322 event = (struct ring_buffer_event *)(bpage->data + e); 4323 4324 switch (event->type_len) { 4325 4326 case RINGBUF_TYPE_TIME_EXTEND: 4327 delta = rb_event_time_stamp(event); 4328 ts += delta; 4329 pr_warn(" 0x%x: [%lld] delta:%lld TIME EXTEND\n", 4330 e, ts, delta); 4331 break; 4332 4333 case RINGBUF_TYPE_TIME_STAMP: 4334 delta = rb_event_time_stamp(event); 4335 ts = rb_fix_abs_ts(delta, ts); 4336 pr_warn(" 0x%x: [%lld] absolute:%lld TIME STAMP\n", 4337 e, ts, delta); 4338 break; 4339 4340 case RINGBUF_TYPE_PADDING: 4341 ts += event->time_delta; 4342 pr_warn(" 0x%x: [%lld] delta:%d PADDING\n", 4343 e, ts, event->time_delta); 4344 break; 4345 4346 case RINGBUF_TYPE_DATA: 4347 ts += event->time_delta; 4348 pr_warn(" 0x%x: [%lld] delta:%d %s%s\n", 4349 e, ts, event->time_delta, 4350 show_flags(event), show_irq(event)); 4351 break; 4352 4353 default: 4354 break; 4355 } 4356 } 4357 pr_warn("expected end:0x%lx last event actually ended at:0x%x\n", tail, e); 4358} 4359 4360static DEFINE_PER_CPU(atomic_t, checking); 4361static atomic_t ts_dump; 4362 4363#define buffer_warn_return(fmt, ...) \ 4364 do { \ 4365 /* If another report is happening, ignore this one */ \ 4366 if (atomic_inc_return(&ts_dump) != 1) { \ 4367 atomic_dec(&ts_dump); \ 4368 goto out; \ 4369 } \ 4370 atomic_inc(&cpu_buffer->record_disabled); \ 4371 pr_warn(fmt, ##__VA_ARGS__); \ 4372 dump_buffer_page(bpage, info, tail); \ 4373 atomic_dec(&ts_dump); \ 4374 /* There's some cases in boot up that this can happen */ \ 4375 if (WARN_ON_ONCE(system_state != SYSTEM_BOOTING)) \ 4376 /* Do not re-enable checking */ \ 4377 return; \ 4378 } while (0) 4379 4380/* 4381 * Check if the current event time stamp matches the deltas on 4382 * the buffer page. 4383 */ 4384static void check_buffer(struct ring_buffer_per_cpu *cpu_buffer, 4385 struct rb_event_info *info, 4386 unsigned long tail) 4387{ 4388 struct buffer_data_page *bpage; 4389 u64 ts, delta; 4390 bool full = false; 4391 int ret; 4392 4393 bpage = info->tail_page->page; 4394 4395 if (tail == CHECK_FULL_PAGE) { 4396 full = true; 4397 tail = local_read(&bpage->commit); 4398 } else if (info->add_timestamp & 4399 (RB_ADD_STAMP_FORCE | RB_ADD_STAMP_ABSOLUTE)) { 4400 /* Ignore events with absolute time stamps */ 4401 return; 4402 } 4403 4404 /* 4405 * Do not check the first event (skip possible extends too). 4406 * Also do not check if previous events have not been committed. 4407 */ 4408 if (tail <= 8 || tail > local_read(&bpage->commit)) 4409 return; 4410 4411 /* 4412 * If this interrupted another event, 4413 */ 4414 if (atomic_inc_return(this_cpu_ptr(&checking)) != 1) 4415 goto out; 4416 4417 ret = rb_read_data_buffer(bpage, tail, cpu_buffer->cpu, &ts, &delta); 4418 if (ret < 0) { 4419 if (delta < ts) { 4420 buffer_warn_return("[CPU: %d]ABSOLUTE TIME WENT BACKWARDS: last ts: %lld absolute ts: %lld\n", 4421 cpu_buffer->cpu, ts, delta); 4422 goto out; 4423 } 4424 } 4425 if ((full && ts > info->ts) || 4426 (!full && ts + info->delta != info->ts)) { 4427 buffer_warn_return("[CPU: %d]TIME DOES NOT MATCH expected:%lld actual:%lld delta:%lld before:%lld after:%lld%s context:%s\n", 4428 cpu_buffer->cpu, 4429 ts + info->delta, info->ts, info->delta, 4430 info->before, info->after, 4431 full ? " (full)" : "", show_interrupt_level()); 4432 } 4433out: 4434 atomic_dec(this_cpu_ptr(&checking)); 4435} 4436#else 4437static inline void check_buffer(struct ring_buffer_per_cpu *cpu_buffer, 4438 struct rb_event_info *info, 4439 unsigned long tail) 4440{ 4441} 4442#endif /* CONFIG_RING_BUFFER_VALIDATE_TIME_DELTAS */ 4443 4444static struct ring_buffer_event * 4445__rb_reserve_next(struct ring_buffer_per_cpu *cpu_buffer, 4446 struct rb_event_info *info) 4447{ 4448 struct ring_buffer_event *event; 4449 struct buffer_page *tail_page; 4450 unsigned long tail, write, w; 4451 4452 /* Don't let the compiler play games with cpu_buffer->tail_page */ 4453 tail_page = info->tail_page = READ_ONCE(cpu_buffer->tail_page); 4454 4455 /*A*/ w = local_read(&tail_page->write) & RB_WRITE_MASK; 4456 barrier(); 4457 rb_time_read(&cpu_buffer->before_stamp, &info->before); 4458 rb_time_read(&cpu_buffer->write_stamp, &info->after); 4459 barrier(); 4460 info->ts = rb_time_stamp(cpu_buffer->buffer); 4461 4462 if ((info->add_timestamp & RB_ADD_STAMP_ABSOLUTE)) { 4463 info->delta = info->ts; 4464 } else { 4465 /* 4466 * If interrupting an event time update, we may need an 4467 * absolute timestamp. 4468 * Don't bother if this is the start of a new page (w == 0). 4469 */ 4470 if (!w) { 4471 /* Use the sub-buffer timestamp */ 4472 info->delta = 0; 4473 } else if (unlikely(info->before != info->after)) { 4474 info->add_timestamp |= RB_ADD_STAMP_FORCE | RB_ADD_STAMP_EXTEND; 4475 info->length += RB_LEN_TIME_EXTEND; 4476 } else { 4477 info->delta = info->ts - info->after; 4478 if (unlikely(test_time_stamp(info->delta))) { 4479 info->add_timestamp |= RB_ADD_STAMP_EXTEND; 4480 info->length += RB_LEN_TIME_EXTEND; 4481 } 4482 } 4483 } 4484 4485 /*B*/ rb_time_set(&cpu_buffer->before_stamp, info->ts); 4486 4487 /*C*/ write = local_add_return(info->length, &tail_page->write); 4488 4489 /* set write to only the index of the write */ 4490 write &= RB_WRITE_MASK; 4491 4492 tail = write - info->length; 4493 4494 /* See if we shot pass the end of this buffer page */ 4495 if (unlikely(write > cpu_buffer->buffer->subbuf_size)) { 4496 check_buffer(cpu_buffer, info, CHECK_FULL_PAGE); 4497 return rb_move_tail(cpu_buffer, tail, info); 4498 } 4499 4500 if (likely(tail == w)) { 4501 /* Nothing interrupted us between A and C */ 4502 /*D*/ rb_time_set(&cpu_buffer->write_stamp, info->ts); 4503 /* 4504 * If something came in between C and D, the write stamp 4505 * may now not be in sync. But that's fine as the before_stamp 4506 * will be different and then next event will just be forced 4507 * to use an absolute timestamp. 4508 */ 4509 if (likely(!(info->add_timestamp & 4510 (RB_ADD_STAMP_FORCE | RB_ADD_STAMP_ABSOLUTE)))) 4511 /* This did not interrupt any time update */ 4512 info->delta = info->ts - info->after; 4513 else 4514 /* Just use full timestamp for interrupting event */ 4515 info->delta = info->ts; 4516 check_buffer(cpu_buffer, info, tail); 4517 } else { 4518 u64 ts; 4519 /* SLOW PATH - Interrupted between A and C */ 4520 4521 /* Save the old before_stamp */ 4522 rb_time_read(&cpu_buffer->before_stamp, &info->before); 4523 4524 /* 4525 * Read a new timestamp and update the before_stamp to make 4526 * the next event after this one force using an absolute 4527 * timestamp. This is in case an interrupt were to come in 4528 * between E and F. 4529 */ 4530 ts = rb_time_stamp(cpu_buffer->buffer); 4531 rb_time_set(&cpu_buffer->before_stamp, ts); 4532 4533 barrier(); 4534 /*E*/ rb_time_read(&cpu_buffer->write_stamp, &info->after); 4535 barrier(); 4536 /*F*/ if (write == (local_read(&tail_page->write) & RB_WRITE_MASK) && 4537 info->after == info->before && info->after < ts) { 4538 /* 4539 * Nothing came after this event between C and F, it is 4540 * safe to use info->after for the delta as it 4541 * matched info->before and is still valid. 4542 */ 4543 info->delta = ts - info->after; 4544 } else { 4545 /* 4546 * Interrupted between C and F: 4547 * Lost the previous events time stamp. Just set the 4548 * delta to zero, and this will be the same time as 4549 * the event this event interrupted. And the events that 4550 * came after this will still be correct (as they would 4551 * have built their delta on the previous event. 4552 */ 4553 info->delta = 0; 4554 } 4555 info->ts = ts; 4556 info->add_timestamp &= ~RB_ADD_STAMP_FORCE; 4557 } 4558 4559 /* 4560 * If this is the first commit on the page, then it has the same 4561 * timestamp as the page itself. 4562 */ 4563 if (unlikely(!tail && !(info->add_timestamp & 4564 (RB_ADD_STAMP_FORCE | RB_ADD_STAMP_ABSOLUTE)))) 4565 info->delta = 0; 4566 4567 /* We reserved something on the buffer */ 4568 4569 event = __rb_page_index(tail_page, tail); 4570 rb_update_event(cpu_buffer, event, info); 4571 4572 local_inc(&tail_page->entries); 4573 4574 /* 4575 * If this is the first commit on the page, then update 4576 * its timestamp. 4577 */ 4578 if (unlikely(!tail)) 4579 tail_page->page->time_stamp = info->ts; 4580 4581 /* account for these added bytes */ 4582 local_add(info->length, &cpu_buffer->entries_bytes); 4583 4584 return event; 4585} 4586 4587static __always_inline struct ring_buffer_event * 4588rb_reserve_next_event(struct trace_buffer *buffer, 4589 struct ring_buffer_per_cpu *cpu_buffer, 4590 unsigned long length) 4591{ 4592 struct ring_buffer_event *event; 4593 struct rb_event_info info; 4594 int nr_loops = 0; 4595 int add_ts_default; 4596 4597 /* 4598 * ring buffer does cmpxchg as well as atomic64 operations 4599 * (which some archs use locking for atomic64), make sure this 4600 * is safe in NMI context 4601 */ 4602 if ((!IS_ENABLED(CONFIG_ARCH_HAVE_NMI_SAFE_CMPXCHG) || 4603 IS_ENABLED(CONFIG_GENERIC_ATOMIC64)) && 4604 (unlikely(in_nmi()))) { 4605 return NULL; 4606 } 4607 4608 rb_start_commit(cpu_buffer); 4609 /* The commit page can not change after this */ 4610 4611#ifdef CONFIG_RING_BUFFER_ALLOW_SWAP 4612 /* 4613 * Due to the ability to swap a cpu buffer from a buffer 4614 * it is possible it was swapped before we committed. 4615 * (committing stops a swap). We check for it here and 4616 * if it happened, we have to fail the write. 4617 */ 4618 barrier(); 4619 if (unlikely(READ_ONCE(cpu_buffer->buffer) != buffer)) { 4620 local_dec(&cpu_buffer->committing); 4621 local_dec(&cpu_buffer->commits); 4622 return NULL; 4623 } 4624#endif 4625 4626 info.length = rb_calculate_event_length(length); 4627 4628 if (ring_buffer_time_stamp_abs(cpu_buffer->buffer)) { 4629 add_ts_default = RB_ADD_STAMP_ABSOLUTE; 4630 info.length += RB_LEN_TIME_EXTEND; 4631 if (info.length > cpu_buffer->buffer->max_data_size) 4632 goto out_fail; 4633 } else { 4634 add_ts_default = RB_ADD_STAMP_NONE; 4635 } 4636 4637 again: 4638 info.add_timestamp = add_ts_default; 4639 info.delta = 0; 4640 4641 /* 4642 * We allow for interrupts to reenter here and do a trace. 4643 * If one does, it will cause this original code to loop 4644 * back here. Even with heavy interrupts happening, this 4645 * should only happen a few times in a row. If this happens 4646 * 1000 times in a row, there must be either an interrupt 4647 * storm or we have something buggy. 4648 * Bail! 4649 */ 4650 if (RB_WARN_ON(cpu_buffer, ++nr_loops > 1000)) 4651 goto out_fail; 4652 4653 event = __rb_reserve_next(cpu_buffer, &info); 4654 4655 if (unlikely(PTR_ERR(event) == -EAGAIN)) { 4656 if (info.add_timestamp & (RB_ADD_STAMP_FORCE | RB_ADD_STAMP_EXTEND)) 4657 info.length -= RB_LEN_TIME_EXTEND; 4658 goto again; 4659 } 4660 4661 if (likely(event)) 4662 return event; 4663 out_fail: 4664 rb_end_commit(cpu_buffer); 4665 return NULL; 4666} 4667 4668/** 4669 * ring_buffer_lock_reserve - reserve a part of the buffer 4670 * @buffer: the ring buffer to reserve from 4671 * @length: the length of the data to reserve (excluding event header) 4672 * 4673 * Returns a reserved event on the ring buffer to copy directly to. 4674 * The user of this interface will need to get the body to write into 4675 * and can use the ring_buffer_event_data() interface. 4676 * 4677 * The length is the length of the data needed, not the event length 4678 * which also includes the event header. 4679 * 4680 * Must be paired with ring_buffer_unlock_commit, unless NULL is returned. 4681 * If NULL is returned, then nothing has been allocated or locked. 4682 */ 4683struct ring_buffer_event * 4684ring_buffer_lock_reserve(struct trace_buffer *buffer, unsigned long length) 4685{ 4686 struct ring_buffer_per_cpu *cpu_buffer; 4687 struct ring_buffer_event *event; 4688 int cpu; 4689 4690 /* If we are tracing schedule, we don't want to recurse */ 4691 preempt_disable_notrace(); 4692 4693 if (unlikely(atomic_read(&buffer->record_disabled))) 4694 goto out; 4695 4696 cpu = raw_smp_processor_id(); 4697 4698 if (unlikely(!cpumask_test_cpu(cpu, buffer->cpumask))) 4699 goto out; 4700 4701 cpu_buffer = buffer->buffers[cpu]; 4702 4703 if (unlikely(atomic_read(&cpu_buffer->record_disabled))) 4704 goto out; 4705 4706 if (unlikely(length > buffer->max_data_size)) 4707 goto out; 4708 4709 if (unlikely(trace_recursive_lock(cpu_buffer))) 4710 goto out; 4711 4712 event = rb_reserve_next_event(buffer, cpu_buffer, length); 4713 if (!event) 4714 goto out_unlock; 4715 4716 return event; 4717 4718 out_unlock: 4719 trace_recursive_unlock(cpu_buffer); 4720 out: 4721 preempt_enable_notrace(); 4722 return NULL; 4723} 4724EXPORT_SYMBOL_GPL(ring_buffer_lock_reserve); 4725 4726/* 4727 * Decrement the entries to the page that an event is on. 4728 * The event does not even need to exist, only the pointer 4729 * to the page it is on. This may only be called before the commit 4730 * takes place. 4731 */ 4732static inline void 4733rb_decrement_entry(struct ring_buffer_per_cpu *cpu_buffer, 4734 struct ring_buffer_event *event) 4735{ 4736 unsigned long addr = (unsigned long)event; 4737 struct buffer_page *bpage = cpu_buffer->commit_page; 4738 struct buffer_page *start; 4739 4740 addr &= ~((PAGE_SIZE << cpu_buffer->buffer->subbuf_order) - 1); 4741 4742 /* Do the likely case first */ 4743 if (likely(bpage->page == (void *)addr)) { 4744 local_dec(&bpage->entries); 4745 return; 4746 } 4747 4748 /* 4749 * Because the commit page may be on the reader page we 4750 * start with the next page and check the end loop there. 4751 */ 4752 rb_inc_page(&bpage); 4753 start = bpage; 4754 do { 4755 if (bpage->page == (void *)addr) { 4756 local_dec(&bpage->entries); 4757 return; 4758 } 4759 rb_inc_page(&bpage); 4760 } while (bpage != start); 4761 4762 /* commit not part of this buffer?? */ 4763 RB_WARN_ON(cpu_buffer, 1); 4764} 4765 4766/** 4767 * ring_buffer_discard_commit - discard an event that has not been committed 4768 * @buffer: the ring buffer 4769 * @event: non committed event to discard 4770 * 4771 * Sometimes an event that is in the ring buffer needs to be ignored. 4772 * This function lets the user discard an event in the ring buffer 4773 * and then that event will not be read later. 4774 * 4775 * This function only works if it is called before the item has been 4776 * committed. It will try to free the event from the ring buffer 4777 * if another event has not been added behind it. 4778 * 4779 * If another event has been added behind it, it will set the event 4780 * up as discarded, and perform the commit. 4781 * 4782 * If this function is called, do not call ring_buffer_unlock_commit on 4783 * the event. 4784 */ 4785void ring_buffer_discard_commit(struct trace_buffer *buffer, 4786 struct ring_buffer_event *event) 4787{ 4788 struct ring_buffer_per_cpu *cpu_buffer; 4789 int cpu; 4790 4791 /* The event is discarded regardless */ 4792 rb_event_discard(event); 4793 4794 cpu = smp_processor_id(); 4795 cpu_buffer = buffer->buffers[cpu]; 4796 4797 /* 4798 * This must only be called if the event has not been 4799 * committed yet. Thus we can assume that preemption 4800 * is still disabled. 4801 */ 4802 RB_WARN_ON(buffer, !local_read(&cpu_buffer->committing)); 4803 4804 rb_decrement_entry(cpu_buffer, event); 4805 rb_try_to_discard(cpu_buffer, event); 4806 rb_end_commit(cpu_buffer); 4807 4808 trace_recursive_unlock(cpu_buffer); 4809 4810 preempt_enable_notrace(); 4811 4812} 4813EXPORT_SYMBOL_GPL(ring_buffer_discard_commit); 4814 4815/** 4816 * ring_buffer_write - write data to the buffer without reserving 4817 * @buffer: The ring buffer to write to. 4818 * @length: The length of the data being written (excluding the event header) 4819 * @data: The data to write to the buffer. 4820 * 4821 * This is like ring_buffer_lock_reserve and ring_buffer_unlock_commit as 4822 * one function. If you already have the data to write to the buffer, it 4823 * may be easier to simply call this function. 4824 * 4825 * Note, like ring_buffer_lock_reserve, the length is the length of the data 4826 * and not the length of the event which would hold the header. 4827 */ 4828int ring_buffer_write(struct trace_buffer *buffer, 4829 unsigned long length, 4830 void *data) 4831{ 4832 struct ring_buffer_per_cpu *cpu_buffer; 4833 struct ring_buffer_event *event; 4834 void *body; 4835 int ret = -EBUSY; 4836 int cpu; 4837 4838 guard(preempt_notrace)(); 4839 4840 if (atomic_read(&buffer->record_disabled)) 4841 return -EBUSY; 4842 4843 cpu = raw_smp_processor_id(); 4844 4845 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 4846 return -EBUSY; 4847 4848 cpu_buffer = buffer->buffers[cpu]; 4849 4850 if (atomic_read(&cpu_buffer->record_disabled)) 4851 return -EBUSY; 4852 4853 if (length > buffer->max_data_size) 4854 return -EBUSY; 4855 4856 if (unlikely(trace_recursive_lock(cpu_buffer))) 4857 return -EBUSY; 4858 4859 event = rb_reserve_next_event(buffer, cpu_buffer, length); 4860 if (!event) 4861 goto out_unlock; 4862 4863 body = rb_event_data(event); 4864 4865 memcpy(body, data, length); 4866 4867 rb_commit(cpu_buffer); 4868 4869 rb_wakeups(buffer, cpu_buffer); 4870 4871 ret = 0; 4872 4873 out_unlock: 4874 trace_recursive_unlock(cpu_buffer); 4875 return ret; 4876} 4877EXPORT_SYMBOL_GPL(ring_buffer_write); 4878 4879/* 4880 * The total entries in the ring buffer is the running counter 4881 * of entries entered into the ring buffer, minus the sum of 4882 * the entries read from the ring buffer and the number of 4883 * entries that were overwritten. 4884 */ 4885static inline unsigned long 4886rb_num_of_entries(struct ring_buffer_per_cpu *cpu_buffer) 4887{ 4888 return local_read(&cpu_buffer->entries) - 4889 (local_read(&cpu_buffer->overrun) + cpu_buffer->read); 4890} 4891 4892static bool rb_per_cpu_empty(struct ring_buffer_per_cpu *cpu_buffer) 4893{ 4894 return !rb_num_of_entries(cpu_buffer); 4895} 4896 4897/** 4898 * ring_buffer_record_disable - stop all writes into the buffer 4899 * @buffer: The ring buffer to stop writes to. 4900 * 4901 * This prevents all writes to the buffer. Any attempt to write 4902 * to the buffer after this will fail and return NULL. 4903 * 4904 * The caller should call synchronize_rcu() after this. 4905 */ 4906void ring_buffer_record_disable(struct trace_buffer *buffer) 4907{ 4908 atomic_inc(&buffer->record_disabled); 4909} 4910EXPORT_SYMBOL_GPL(ring_buffer_record_disable); 4911 4912/** 4913 * ring_buffer_record_enable - enable writes to the buffer 4914 * @buffer: The ring buffer to enable writes 4915 * 4916 * Note, multiple disables will need the same number of enables 4917 * to truly enable the writing (much like preempt_disable). 4918 */ 4919void ring_buffer_record_enable(struct trace_buffer *buffer) 4920{ 4921 atomic_dec(&buffer->record_disabled); 4922} 4923EXPORT_SYMBOL_GPL(ring_buffer_record_enable); 4924 4925/** 4926 * ring_buffer_record_off - stop all writes into the buffer 4927 * @buffer: The ring buffer to stop writes to. 4928 * 4929 * This prevents all writes to the buffer. Any attempt to write 4930 * to the buffer after this will fail and return NULL. 4931 * 4932 * This is different than ring_buffer_record_disable() as 4933 * it works like an on/off switch, where as the disable() version 4934 * must be paired with a enable(). 4935 */ 4936void ring_buffer_record_off(struct trace_buffer *buffer) 4937{ 4938 unsigned int rd; 4939 unsigned int new_rd; 4940 4941 rd = atomic_read(&buffer->record_disabled); 4942 do { 4943 new_rd = rd | RB_BUFFER_OFF; 4944 } while (!atomic_try_cmpxchg(&buffer->record_disabled, &rd, new_rd)); 4945} 4946EXPORT_SYMBOL_GPL(ring_buffer_record_off); 4947 4948/** 4949 * ring_buffer_record_on - restart writes into the buffer 4950 * @buffer: The ring buffer to start writes to. 4951 * 4952 * This enables all writes to the buffer that was disabled by 4953 * ring_buffer_record_off(). 4954 * 4955 * This is different than ring_buffer_record_enable() as 4956 * it works like an on/off switch, where as the enable() version 4957 * must be paired with a disable(). 4958 */ 4959void ring_buffer_record_on(struct trace_buffer *buffer) 4960{ 4961 unsigned int rd; 4962 unsigned int new_rd; 4963 4964 rd = atomic_read(&buffer->record_disabled); 4965 do { 4966 new_rd = rd & ~RB_BUFFER_OFF; 4967 } while (!atomic_try_cmpxchg(&buffer->record_disabled, &rd, new_rd)); 4968} 4969EXPORT_SYMBOL_GPL(ring_buffer_record_on); 4970 4971/** 4972 * ring_buffer_record_is_on - return true if the ring buffer can write 4973 * @buffer: The ring buffer to see if write is enabled 4974 * 4975 * Returns true if the ring buffer is in a state that it accepts writes. 4976 */ 4977bool ring_buffer_record_is_on(struct trace_buffer *buffer) 4978{ 4979 return !atomic_read(&buffer->record_disabled); 4980} 4981 4982/** 4983 * ring_buffer_record_is_set_on - return true if the ring buffer is set writable 4984 * @buffer: The ring buffer to see if write is set enabled 4985 * 4986 * Returns true if the ring buffer is set writable by ring_buffer_record_on(). 4987 * Note that this does NOT mean it is in a writable state. 4988 * 4989 * It may return true when the ring buffer has been disabled by 4990 * ring_buffer_record_disable(), as that is a temporary disabling of 4991 * the ring buffer. 4992 */ 4993bool ring_buffer_record_is_set_on(struct trace_buffer *buffer) 4994{ 4995 return !(atomic_read(&buffer->record_disabled) & RB_BUFFER_OFF); 4996} 4997 4998/** 4999 * ring_buffer_record_is_on_cpu - return true if the ring buffer can write 5000 * @buffer: The ring buffer to see if write is enabled 5001 * @cpu: The CPU to test if the ring buffer can write too 5002 * 5003 * Returns true if the ring buffer is in a state that it accepts writes 5004 * for a particular CPU. 5005 */ 5006bool ring_buffer_record_is_on_cpu(struct trace_buffer *buffer, int cpu) 5007{ 5008 struct ring_buffer_per_cpu *cpu_buffer; 5009 5010 cpu_buffer = buffer->buffers[cpu]; 5011 5012 return ring_buffer_record_is_set_on(buffer) && 5013 !atomic_read(&cpu_buffer->record_disabled); 5014} 5015 5016/** 5017 * ring_buffer_record_disable_cpu - stop all writes into the cpu_buffer 5018 * @buffer: The ring buffer to stop writes to. 5019 * @cpu: The CPU buffer to stop 5020 * 5021 * This prevents all writes to the buffer. Any attempt to write 5022 * to the buffer after this will fail and return NULL. 5023 * 5024 * The caller should call synchronize_rcu() after this. 5025 */ 5026void ring_buffer_record_disable_cpu(struct trace_buffer *buffer, int cpu) 5027{ 5028 struct ring_buffer_per_cpu *cpu_buffer; 5029 5030 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 5031 return; 5032 5033 cpu_buffer = buffer->buffers[cpu]; 5034 atomic_inc(&cpu_buffer->record_disabled); 5035} 5036EXPORT_SYMBOL_GPL(ring_buffer_record_disable_cpu); 5037 5038/** 5039 * ring_buffer_record_enable_cpu - enable writes to the buffer 5040 * @buffer: The ring buffer to enable writes 5041 * @cpu: The CPU to enable. 5042 * 5043 * Note, multiple disables will need the same number of enables 5044 * to truly enable the writing (much like preempt_disable). 5045 */ 5046void ring_buffer_record_enable_cpu(struct trace_buffer *buffer, int cpu) 5047{ 5048 struct ring_buffer_per_cpu *cpu_buffer; 5049 5050 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 5051 return; 5052 5053 cpu_buffer = buffer->buffers[cpu]; 5054 atomic_dec(&cpu_buffer->record_disabled); 5055} 5056EXPORT_SYMBOL_GPL(ring_buffer_record_enable_cpu); 5057 5058/** 5059 * ring_buffer_oldest_event_ts - get the oldest event timestamp from the buffer 5060 * @buffer: The ring buffer 5061 * @cpu: The per CPU buffer to read from. 5062 */ 5063u64 ring_buffer_oldest_event_ts(struct trace_buffer *buffer, int cpu) 5064{ 5065 unsigned long flags; 5066 struct ring_buffer_per_cpu *cpu_buffer; 5067 struct buffer_page *bpage; 5068 u64 ret = 0; 5069 5070 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 5071 return 0; 5072 5073 cpu_buffer = buffer->buffers[cpu]; 5074 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 5075 /* 5076 * if the tail is on reader_page, oldest time stamp is on the reader 5077 * page 5078 */ 5079 if (cpu_buffer->tail_page == cpu_buffer->reader_page) 5080 bpage = cpu_buffer->reader_page; 5081 else 5082 bpage = rb_set_head_page(cpu_buffer); 5083 if (bpage) 5084 ret = bpage->page->time_stamp; 5085 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 5086 5087 return ret; 5088} 5089EXPORT_SYMBOL_GPL(ring_buffer_oldest_event_ts); 5090 5091/** 5092 * ring_buffer_bytes_cpu - get the number of bytes unconsumed in a cpu buffer 5093 * @buffer: The ring buffer 5094 * @cpu: The per CPU buffer to read from. 5095 */ 5096unsigned long ring_buffer_bytes_cpu(struct trace_buffer *buffer, int cpu) 5097{ 5098 struct ring_buffer_per_cpu *cpu_buffer; 5099 unsigned long ret; 5100 5101 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 5102 return 0; 5103 5104 cpu_buffer = buffer->buffers[cpu]; 5105 ret = local_read(&cpu_buffer->entries_bytes) - cpu_buffer->read_bytes; 5106 5107 return ret; 5108} 5109EXPORT_SYMBOL_GPL(ring_buffer_bytes_cpu); 5110 5111/** 5112 * ring_buffer_entries_cpu - get the number of entries in a cpu buffer 5113 * @buffer: The ring buffer 5114 * @cpu: The per CPU buffer to get the entries from. 5115 */ 5116unsigned long ring_buffer_entries_cpu(struct trace_buffer *buffer, int cpu) 5117{ 5118 struct ring_buffer_per_cpu *cpu_buffer; 5119 5120 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 5121 return 0; 5122 5123 cpu_buffer = buffer->buffers[cpu]; 5124 5125 return rb_num_of_entries(cpu_buffer); 5126} 5127EXPORT_SYMBOL_GPL(ring_buffer_entries_cpu); 5128 5129/** 5130 * ring_buffer_overrun_cpu - get the number of overruns caused by the ring 5131 * buffer wrapping around (only if RB_FL_OVERWRITE is on). 5132 * @buffer: The ring buffer 5133 * @cpu: The per CPU buffer to get the number of overruns from 5134 */ 5135unsigned long ring_buffer_overrun_cpu(struct trace_buffer *buffer, int cpu) 5136{ 5137 struct ring_buffer_per_cpu *cpu_buffer; 5138 unsigned long ret; 5139 5140 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 5141 return 0; 5142 5143 cpu_buffer = buffer->buffers[cpu]; 5144 ret = local_read(&cpu_buffer->overrun); 5145 5146 return ret; 5147} 5148EXPORT_SYMBOL_GPL(ring_buffer_overrun_cpu); 5149 5150/** 5151 * ring_buffer_commit_overrun_cpu - get the number of overruns caused by 5152 * commits failing due to the buffer wrapping around while there are uncommitted 5153 * events, such as during an interrupt storm. 5154 * @buffer: The ring buffer 5155 * @cpu: The per CPU buffer to get the number of overruns from 5156 */ 5157unsigned long 5158ring_buffer_commit_overrun_cpu(struct trace_buffer *buffer, int cpu) 5159{ 5160 struct ring_buffer_per_cpu *cpu_buffer; 5161 unsigned long ret; 5162 5163 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 5164 return 0; 5165 5166 cpu_buffer = buffer->buffers[cpu]; 5167 ret = local_read(&cpu_buffer->commit_overrun); 5168 5169 return ret; 5170} 5171EXPORT_SYMBOL_GPL(ring_buffer_commit_overrun_cpu); 5172 5173/** 5174 * ring_buffer_dropped_events_cpu - get the number of dropped events caused by 5175 * the ring buffer filling up (only if RB_FL_OVERWRITE is off). 5176 * @buffer: The ring buffer 5177 * @cpu: The per CPU buffer to get the number of overruns from 5178 */ 5179unsigned long 5180ring_buffer_dropped_events_cpu(struct trace_buffer *buffer, int cpu) 5181{ 5182 struct ring_buffer_per_cpu *cpu_buffer; 5183 unsigned long ret; 5184 5185 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 5186 return 0; 5187 5188 cpu_buffer = buffer->buffers[cpu]; 5189 ret = local_read(&cpu_buffer->dropped_events); 5190 5191 return ret; 5192} 5193EXPORT_SYMBOL_GPL(ring_buffer_dropped_events_cpu); 5194 5195/** 5196 * ring_buffer_read_events_cpu - get the number of events successfully read 5197 * @buffer: The ring buffer 5198 * @cpu: The per CPU buffer to get the number of events read 5199 */ 5200unsigned long 5201ring_buffer_read_events_cpu(struct trace_buffer *buffer, int cpu) 5202{ 5203 struct ring_buffer_per_cpu *cpu_buffer; 5204 5205 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 5206 return 0; 5207 5208 cpu_buffer = buffer->buffers[cpu]; 5209 return cpu_buffer->read; 5210} 5211EXPORT_SYMBOL_GPL(ring_buffer_read_events_cpu); 5212 5213/** 5214 * ring_buffer_entries - get the number of entries in a buffer 5215 * @buffer: The ring buffer 5216 * 5217 * Returns the total number of entries in the ring buffer 5218 * (all CPU entries) 5219 */ 5220unsigned long ring_buffer_entries(struct trace_buffer *buffer) 5221{ 5222 struct ring_buffer_per_cpu *cpu_buffer; 5223 unsigned long entries = 0; 5224 int cpu; 5225 5226 /* if you care about this being correct, lock the buffer */ 5227 for_each_buffer_cpu(buffer, cpu) { 5228 cpu_buffer = buffer->buffers[cpu]; 5229 entries += rb_num_of_entries(cpu_buffer); 5230 } 5231 5232 return entries; 5233} 5234EXPORT_SYMBOL_GPL(ring_buffer_entries); 5235 5236/** 5237 * ring_buffer_overruns - get the number of overruns in buffer 5238 * @buffer: The ring buffer 5239 * 5240 * Returns the total number of overruns in the ring buffer 5241 * (all CPU entries) 5242 */ 5243unsigned long ring_buffer_overruns(struct trace_buffer *buffer) 5244{ 5245 struct ring_buffer_per_cpu *cpu_buffer; 5246 unsigned long overruns = 0; 5247 int cpu; 5248 5249 /* if you care about this being correct, lock the buffer */ 5250 for_each_buffer_cpu(buffer, cpu) { 5251 cpu_buffer = buffer->buffers[cpu]; 5252 overruns += local_read(&cpu_buffer->overrun); 5253 } 5254 5255 return overruns; 5256} 5257EXPORT_SYMBOL_GPL(ring_buffer_overruns); 5258 5259static void rb_iter_reset(struct ring_buffer_iter *iter) 5260{ 5261 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer; 5262 5263 /* Iterator usage is expected to have record disabled */ 5264 iter->head_page = cpu_buffer->reader_page; 5265 iter->head = cpu_buffer->reader_page->read; 5266 iter->next_event = iter->head; 5267 5268 iter->cache_reader_page = iter->head_page; 5269 iter->cache_read = cpu_buffer->read; 5270 iter->cache_pages_removed = cpu_buffer->pages_removed; 5271 5272 if (iter->head) { 5273 iter->read_stamp = cpu_buffer->read_stamp; 5274 iter->page_stamp = cpu_buffer->reader_page->page->time_stamp; 5275 } else { 5276 iter->read_stamp = iter->head_page->page->time_stamp; 5277 iter->page_stamp = iter->read_stamp; 5278 } 5279} 5280 5281/** 5282 * ring_buffer_iter_reset - reset an iterator 5283 * @iter: The iterator to reset 5284 * 5285 * Resets the iterator, so that it will start from the beginning 5286 * again. 5287 */ 5288void ring_buffer_iter_reset(struct ring_buffer_iter *iter) 5289{ 5290 struct ring_buffer_per_cpu *cpu_buffer; 5291 unsigned long flags; 5292 5293 if (!iter) 5294 return; 5295 5296 cpu_buffer = iter->cpu_buffer; 5297 5298 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 5299 rb_iter_reset(iter); 5300 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 5301} 5302EXPORT_SYMBOL_GPL(ring_buffer_iter_reset); 5303 5304/** 5305 * ring_buffer_iter_empty - check if an iterator has no more to read 5306 * @iter: The iterator to check 5307 */ 5308int ring_buffer_iter_empty(struct ring_buffer_iter *iter) 5309{ 5310 struct ring_buffer_per_cpu *cpu_buffer; 5311 struct buffer_page *reader; 5312 struct buffer_page *head_page; 5313 struct buffer_page *commit_page; 5314 struct buffer_page *curr_commit_page; 5315 unsigned commit; 5316 u64 curr_commit_ts; 5317 u64 commit_ts; 5318 5319 cpu_buffer = iter->cpu_buffer; 5320 reader = cpu_buffer->reader_page; 5321 head_page = cpu_buffer->head_page; 5322 commit_page = READ_ONCE(cpu_buffer->commit_page); 5323 commit_ts = commit_page->page->time_stamp; 5324 5325 /* 5326 * When the writer goes across pages, it issues a cmpxchg which 5327 * is a mb(), which will synchronize with the rmb here. 5328 * (see rb_tail_page_update()) 5329 */ 5330 smp_rmb(); 5331 commit = rb_page_commit(commit_page); 5332 /* We want to make sure that the commit page doesn't change */ 5333 smp_rmb(); 5334 5335 /* Make sure commit page didn't change */ 5336 curr_commit_page = READ_ONCE(cpu_buffer->commit_page); 5337 curr_commit_ts = READ_ONCE(curr_commit_page->page->time_stamp); 5338 5339 /* If the commit page changed, then there's more data */ 5340 if (curr_commit_page != commit_page || 5341 curr_commit_ts != commit_ts) 5342 return 0; 5343 5344 /* Still racy, as it may return a false positive, but that's OK */ 5345 return ((iter->head_page == commit_page && iter->head >= commit) || 5346 (iter->head_page == reader && commit_page == head_page && 5347 head_page->read == commit && 5348 iter->head == rb_page_size(cpu_buffer->reader_page))); 5349} 5350EXPORT_SYMBOL_GPL(ring_buffer_iter_empty); 5351 5352static void 5353rb_update_read_stamp(struct ring_buffer_per_cpu *cpu_buffer, 5354 struct ring_buffer_event *event) 5355{ 5356 u64 delta; 5357 5358 switch (event->type_len) { 5359 case RINGBUF_TYPE_PADDING: 5360 return; 5361 5362 case RINGBUF_TYPE_TIME_EXTEND: 5363 delta = rb_event_time_stamp(event); 5364 cpu_buffer->read_stamp += delta; 5365 return; 5366 5367 case RINGBUF_TYPE_TIME_STAMP: 5368 delta = rb_event_time_stamp(event); 5369 delta = rb_fix_abs_ts(delta, cpu_buffer->read_stamp); 5370 cpu_buffer->read_stamp = delta; 5371 return; 5372 5373 case RINGBUF_TYPE_DATA: 5374 cpu_buffer->read_stamp += event->time_delta; 5375 return; 5376 5377 default: 5378 RB_WARN_ON(cpu_buffer, 1); 5379 } 5380} 5381 5382static void 5383rb_update_iter_read_stamp(struct ring_buffer_iter *iter, 5384 struct ring_buffer_event *event) 5385{ 5386 u64 delta; 5387 5388 switch (event->type_len) { 5389 case RINGBUF_TYPE_PADDING: 5390 return; 5391 5392 case RINGBUF_TYPE_TIME_EXTEND: 5393 delta = rb_event_time_stamp(event); 5394 iter->read_stamp += delta; 5395 return; 5396 5397 case RINGBUF_TYPE_TIME_STAMP: 5398 delta = rb_event_time_stamp(event); 5399 delta = rb_fix_abs_ts(delta, iter->read_stamp); 5400 iter->read_stamp = delta; 5401 return; 5402 5403 case RINGBUF_TYPE_DATA: 5404 iter->read_stamp += event->time_delta; 5405 return; 5406 5407 default: 5408 RB_WARN_ON(iter->cpu_buffer, 1); 5409 } 5410} 5411 5412static struct buffer_page * 5413rb_get_reader_page(struct ring_buffer_per_cpu *cpu_buffer) 5414{ 5415 struct buffer_page *reader = NULL; 5416 unsigned long bsize = READ_ONCE(cpu_buffer->buffer->subbuf_size); 5417 unsigned long overwrite; 5418 unsigned long flags; 5419 int nr_loops = 0; 5420 bool ret; 5421 5422 local_irq_save(flags); 5423 arch_spin_lock(&cpu_buffer->lock); 5424 5425 again: 5426 /* 5427 * This should normally only loop twice. But because the 5428 * start of the reader inserts an empty page, it causes 5429 * a case where we will loop three times. There should be no 5430 * reason to loop four times (that I know of). 5431 */ 5432 if (RB_WARN_ON(cpu_buffer, ++nr_loops > 3)) { 5433 reader = NULL; 5434 goto out; 5435 } 5436 5437 reader = cpu_buffer->reader_page; 5438 5439 /* If there's more to read, return this page */ 5440 if (cpu_buffer->reader_page->read < rb_page_size(reader)) 5441 goto out; 5442 5443 /* Never should we have an index greater than the size */ 5444 if (RB_WARN_ON(cpu_buffer, 5445 cpu_buffer->reader_page->read > rb_page_size(reader))) 5446 goto out; 5447 5448 /* check if we caught up to the tail */ 5449 reader = NULL; 5450 if (cpu_buffer->commit_page == cpu_buffer->reader_page) 5451 goto out; 5452 5453 /* Don't bother swapping if the ring buffer is empty */ 5454 if (rb_num_of_entries(cpu_buffer) == 0) 5455 goto out; 5456 5457 /* 5458 * Reset the reader page to size zero. 5459 */ 5460 local_set(&cpu_buffer->reader_page->write, 0); 5461 local_set(&cpu_buffer->reader_page->entries, 0); 5462 cpu_buffer->reader_page->real_end = 0; 5463 5464 spin: 5465 /* 5466 * Splice the empty reader page into the list around the head. 5467 */ 5468 reader = rb_set_head_page(cpu_buffer); 5469 if (!reader) 5470 goto out; 5471 cpu_buffer->reader_page->list.next = rb_list_head(reader->list.next); 5472 cpu_buffer->reader_page->list.prev = reader->list.prev; 5473 5474 /* 5475 * cpu_buffer->pages just needs to point to the buffer, it 5476 * has no specific buffer page to point to. Lets move it out 5477 * of our way so we don't accidentally swap it. 5478 */ 5479 cpu_buffer->pages = reader->list.prev; 5480 5481 /* The reader page will be pointing to the new head */ 5482 rb_set_list_to_head(&cpu_buffer->reader_page->list); 5483 5484 /* 5485 * We want to make sure we read the overruns after we set up our 5486 * pointers to the next object. The writer side does a 5487 * cmpxchg to cross pages which acts as the mb on the writer 5488 * side. Note, the reader will constantly fail the swap 5489 * while the writer is updating the pointers, so this 5490 * guarantees that the overwrite recorded here is the one we 5491 * want to compare with the last_overrun. 5492 */ 5493 smp_mb(); 5494 overwrite = local_read(&(cpu_buffer->overrun)); 5495 5496 /* 5497 * Here's the tricky part. 5498 * 5499 * We need to move the pointer past the header page. 5500 * But we can only do that if a writer is not currently 5501 * moving it. The page before the header page has the 5502 * flag bit '1' set if it is pointing to the page we want. 5503 * but if the writer is in the process of moving it 5504 * then it will be '2' or already moved '0'. 5505 */ 5506 5507 ret = rb_head_page_replace(reader, cpu_buffer->reader_page); 5508 5509 /* 5510 * If we did not convert it, then we must try again. 5511 */ 5512 if (!ret) 5513 goto spin; 5514 5515 if (cpu_buffer->ring_meta) 5516 rb_update_meta_reader(cpu_buffer, reader); 5517 5518 /* 5519 * Yay! We succeeded in replacing the page. 5520 * 5521 * Now make the new head point back to the reader page. 5522 */ 5523 rb_list_head(reader->list.next)->prev = &cpu_buffer->reader_page->list; 5524 rb_inc_page(&cpu_buffer->head_page); 5525 5526 cpu_buffer->cnt++; 5527 local_inc(&cpu_buffer->pages_read); 5528 5529 /* Finally update the reader page to the new head */ 5530 cpu_buffer->reader_page = reader; 5531 cpu_buffer->reader_page->read = 0; 5532 5533 if (overwrite != cpu_buffer->last_overrun) { 5534 cpu_buffer->lost_events = overwrite - cpu_buffer->last_overrun; 5535 cpu_buffer->last_overrun = overwrite; 5536 } 5537 5538 goto again; 5539 5540 out: 5541 /* Update the read_stamp on the first event */ 5542 if (reader && reader->read == 0) 5543 cpu_buffer->read_stamp = reader->page->time_stamp; 5544 5545 arch_spin_unlock(&cpu_buffer->lock); 5546 local_irq_restore(flags); 5547 5548 /* 5549 * The writer has preempt disable, wait for it. But not forever 5550 * Although, 1 second is pretty much "forever" 5551 */ 5552#define USECS_WAIT 1000000 5553 for (nr_loops = 0; nr_loops < USECS_WAIT; nr_loops++) { 5554 /* If the write is past the end of page, a writer is still updating it */ 5555 if (likely(!reader || rb_page_write(reader) <= bsize)) 5556 break; 5557 5558 udelay(1); 5559 5560 /* Get the latest version of the reader write value */ 5561 smp_rmb(); 5562 } 5563 5564 /* The writer is not moving forward? Something is wrong */ 5565 if (RB_WARN_ON(cpu_buffer, nr_loops == USECS_WAIT)) 5566 reader = NULL; 5567 5568 /* 5569 * Make sure we see any padding after the write update 5570 * (see rb_reset_tail()). 5571 * 5572 * In addition, a writer may be writing on the reader page 5573 * if the page has not been fully filled, so the read barrier 5574 * is also needed to make sure we see the content of what is 5575 * committed by the writer (see rb_set_commit_to_write()). 5576 */ 5577 smp_rmb(); 5578 5579 5580 return reader; 5581} 5582 5583static void rb_advance_reader(struct ring_buffer_per_cpu *cpu_buffer) 5584{ 5585 struct ring_buffer_event *event; 5586 struct buffer_page *reader; 5587 unsigned length; 5588 5589 reader = rb_get_reader_page(cpu_buffer); 5590 5591 /* This function should not be called when buffer is empty */ 5592 if (RB_WARN_ON(cpu_buffer, !reader)) 5593 return; 5594 5595 event = rb_reader_event(cpu_buffer); 5596 5597 if (event->type_len <= RINGBUF_TYPE_DATA_TYPE_LEN_MAX) 5598 cpu_buffer->read++; 5599 5600 rb_update_read_stamp(cpu_buffer, event); 5601 5602 length = rb_event_length(event); 5603 cpu_buffer->reader_page->read += length; 5604 cpu_buffer->read_bytes += length; 5605} 5606 5607static void rb_advance_iter(struct ring_buffer_iter *iter) 5608{ 5609 struct ring_buffer_per_cpu *cpu_buffer; 5610 5611 cpu_buffer = iter->cpu_buffer; 5612 5613 /* If head == next_event then we need to jump to the next event */ 5614 if (iter->head == iter->next_event) { 5615 /* If the event gets overwritten again, there's nothing to do */ 5616 if (rb_iter_head_event(iter) == NULL) 5617 return; 5618 } 5619 5620 iter->head = iter->next_event; 5621 5622 /* 5623 * Check if we are at the end of the buffer. 5624 */ 5625 if (iter->next_event >= rb_page_size(iter->head_page)) { 5626 /* discarded commits can make the page empty */ 5627 if (iter->head_page == cpu_buffer->commit_page) 5628 return; 5629 rb_inc_iter(iter); 5630 return; 5631 } 5632 5633 rb_update_iter_read_stamp(iter, iter->event); 5634} 5635 5636static int rb_lost_events(struct ring_buffer_per_cpu *cpu_buffer) 5637{ 5638 return cpu_buffer->lost_events; 5639} 5640 5641static struct ring_buffer_event * 5642rb_buffer_peek(struct ring_buffer_per_cpu *cpu_buffer, u64 *ts, 5643 unsigned long *lost_events) 5644{ 5645 struct ring_buffer_event *event; 5646 struct buffer_page *reader; 5647 int nr_loops = 0; 5648 5649 if (ts) 5650 *ts = 0; 5651 again: 5652 /* 5653 * We repeat when a time extend is encountered. 5654 * Since the time extend is always attached to a data event, 5655 * we should never loop more than once. 5656 * (We never hit the following condition more than twice). 5657 */ 5658 if (RB_WARN_ON(cpu_buffer, ++nr_loops > 2)) 5659 return NULL; 5660 5661 reader = rb_get_reader_page(cpu_buffer); 5662 if (!reader) 5663 return NULL; 5664 5665 event = rb_reader_event(cpu_buffer); 5666 5667 switch (event->type_len) { 5668 case RINGBUF_TYPE_PADDING: 5669 if (rb_null_event(event)) 5670 RB_WARN_ON(cpu_buffer, 1); 5671 /* 5672 * Because the writer could be discarding every 5673 * event it creates (which would probably be bad) 5674 * if we were to go back to "again" then we may never 5675 * catch up, and will trigger the warn on, or lock 5676 * the box. Return the padding, and we will release 5677 * the current locks, and try again. 5678 */ 5679 return event; 5680 5681 case RINGBUF_TYPE_TIME_EXTEND: 5682 /* Internal data, OK to advance */ 5683 rb_advance_reader(cpu_buffer); 5684 goto again; 5685 5686 case RINGBUF_TYPE_TIME_STAMP: 5687 if (ts) { 5688 *ts = rb_event_time_stamp(event); 5689 *ts = rb_fix_abs_ts(*ts, reader->page->time_stamp); 5690 ring_buffer_normalize_time_stamp(cpu_buffer->buffer, 5691 cpu_buffer->cpu, ts); 5692 } 5693 /* Internal data, OK to advance */ 5694 rb_advance_reader(cpu_buffer); 5695 goto again; 5696 5697 case RINGBUF_TYPE_DATA: 5698 if (ts && !(*ts)) { 5699 *ts = cpu_buffer->read_stamp + event->time_delta; 5700 ring_buffer_normalize_time_stamp(cpu_buffer->buffer, 5701 cpu_buffer->cpu, ts); 5702 } 5703 if (lost_events) 5704 *lost_events = rb_lost_events(cpu_buffer); 5705 return event; 5706 5707 default: 5708 RB_WARN_ON(cpu_buffer, 1); 5709 } 5710 5711 return NULL; 5712} 5713EXPORT_SYMBOL_GPL(ring_buffer_peek); 5714 5715static struct ring_buffer_event * 5716rb_iter_peek(struct ring_buffer_iter *iter, u64 *ts) 5717{ 5718 struct trace_buffer *buffer; 5719 struct ring_buffer_per_cpu *cpu_buffer; 5720 struct ring_buffer_event *event; 5721 int nr_loops = 0; 5722 5723 if (ts) 5724 *ts = 0; 5725 5726 cpu_buffer = iter->cpu_buffer; 5727 buffer = cpu_buffer->buffer; 5728 5729 /* 5730 * Check if someone performed a consuming read to the buffer 5731 * or removed some pages from the buffer. In these cases, 5732 * iterator was invalidated and we need to reset it. 5733 */ 5734 if (unlikely(iter->cache_read != cpu_buffer->read || 5735 iter->cache_reader_page != cpu_buffer->reader_page || 5736 iter->cache_pages_removed != cpu_buffer->pages_removed)) 5737 rb_iter_reset(iter); 5738 5739 again: 5740 if (ring_buffer_iter_empty(iter)) 5741 return NULL; 5742 5743 /* 5744 * As the writer can mess with what the iterator is trying 5745 * to read, just give up if we fail to get an event after 5746 * three tries. The iterator is not as reliable when reading 5747 * the ring buffer with an active write as the consumer is. 5748 * Do not warn if the three failures is reached. 5749 */ 5750 if (++nr_loops > 3) 5751 return NULL; 5752 5753 if (rb_per_cpu_empty(cpu_buffer)) 5754 return NULL; 5755 5756 if (iter->head >= rb_page_size(iter->head_page)) { 5757 rb_inc_iter(iter); 5758 goto again; 5759 } 5760 5761 event = rb_iter_head_event(iter); 5762 if (!event) 5763 goto again; 5764 5765 switch (event->type_len) { 5766 case RINGBUF_TYPE_PADDING: 5767 if (rb_null_event(event)) { 5768 rb_inc_iter(iter); 5769 goto again; 5770 } 5771 rb_advance_iter(iter); 5772 return event; 5773 5774 case RINGBUF_TYPE_TIME_EXTEND: 5775 /* Internal data, OK to advance */ 5776 rb_advance_iter(iter); 5777 goto again; 5778 5779 case RINGBUF_TYPE_TIME_STAMP: 5780 if (ts) { 5781 *ts = rb_event_time_stamp(event); 5782 *ts = rb_fix_abs_ts(*ts, iter->head_page->page->time_stamp); 5783 ring_buffer_normalize_time_stamp(cpu_buffer->buffer, 5784 cpu_buffer->cpu, ts); 5785 } 5786 /* Internal data, OK to advance */ 5787 rb_advance_iter(iter); 5788 goto again; 5789 5790 case RINGBUF_TYPE_DATA: 5791 if (ts && !(*ts)) { 5792 *ts = iter->read_stamp + event->time_delta; 5793 ring_buffer_normalize_time_stamp(buffer, 5794 cpu_buffer->cpu, ts); 5795 } 5796 return event; 5797 5798 default: 5799 RB_WARN_ON(cpu_buffer, 1); 5800 } 5801 5802 return NULL; 5803} 5804EXPORT_SYMBOL_GPL(ring_buffer_iter_peek); 5805 5806static inline bool rb_reader_lock(struct ring_buffer_per_cpu *cpu_buffer) 5807{ 5808 if (likely(!in_nmi())) { 5809 raw_spin_lock(&cpu_buffer->reader_lock); 5810 return true; 5811 } 5812 5813 /* 5814 * If an NMI die dumps out the content of the ring buffer 5815 * trylock must be used to prevent a deadlock if the NMI 5816 * preempted a task that holds the ring buffer locks. If 5817 * we get the lock then all is fine, if not, then continue 5818 * to do the read, but this can corrupt the ring buffer, 5819 * so it must be permanently disabled from future writes. 5820 * Reading from NMI is a oneshot deal. 5821 */ 5822 if (raw_spin_trylock(&cpu_buffer->reader_lock)) 5823 return true; 5824 5825 /* Continue without locking, but disable the ring buffer */ 5826 atomic_inc(&cpu_buffer->record_disabled); 5827 return false; 5828} 5829 5830static inline void 5831rb_reader_unlock(struct ring_buffer_per_cpu *cpu_buffer, bool locked) 5832{ 5833 if (likely(locked)) 5834 raw_spin_unlock(&cpu_buffer->reader_lock); 5835} 5836 5837/** 5838 * ring_buffer_peek - peek at the next event to be read 5839 * @buffer: The ring buffer to read 5840 * @cpu: The cpu to peak at 5841 * @ts: The timestamp counter of this event. 5842 * @lost_events: a variable to store if events were lost (may be NULL) 5843 * 5844 * This will return the event that will be read next, but does 5845 * not consume the data. 5846 */ 5847struct ring_buffer_event * 5848ring_buffer_peek(struct trace_buffer *buffer, int cpu, u64 *ts, 5849 unsigned long *lost_events) 5850{ 5851 struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu]; 5852 struct ring_buffer_event *event; 5853 unsigned long flags; 5854 bool dolock; 5855 5856 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 5857 return NULL; 5858 5859 again: 5860 local_irq_save(flags); 5861 dolock = rb_reader_lock(cpu_buffer); 5862 event = rb_buffer_peek(cpu_buffer, ts, lost_events); 5863 if (event && event->type_len == RINGBUF_TYPE_PADDING) 5864 rb_advance_reader(cpu_buffer); 5865 rb_reader_unlock(cpu_buffer, dolock); 5866 local_irq_restore(flags); 5867 5868 if (event && event->type_len == RINGBUF_TYPE_PADDING) 5869 goto again; 5870 5871 return event; 5872} 5873 5874/** ring_buffer_iter_dropped - report if there are dropped events 5875 * @iter: The ring buffer iterator 5876 * 5877 * Returns true if there was dropped events since the last peek. 5878 */ 5879bool ring_buffer_iter_dropped(struct ring_buffer_iter *iter) 5880{ 5881 bool ret = iter->missed_events != 0; 5882 5883 iter->missed_events = 0; 5884 return ret; 5885} 5886EXPORT_SYMBOL_GPL(ring_buffer_iter_dropped); 5887 5888/** 5889 * ring_buffer_iter_peek - peek at the next event to be read 5890 * @iter: The ring buffer iterator 5891 * @ts: The timestamp counter of this event. 5892 * 5893 * This will return the event that will be read next, but does 5894 * not increment the iterator. 5895 */ 5896struct ring_buffer_event * 5897ring_buffer_iter_peek(struct ring_buffer_iter *iter, u64 *ts) 5898{ 5899 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer; 5900 struct ring_buffer_event *event; 5901 unsigned long flags; 5902 5903 again: 5904 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 5905 event = rb_iter_peek(iter, ts); 5906 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 5907 5908 if (event && event->type_len == RINGBUF_TYPE_PADDING) 5909 goto again; 5910 5911 return event; 5912} 5913 5914/** 5915 * ring_buffer_consume - return an event and consume it 5916 * @buffer: The ring buffer to get the next event from 5917 * @cpu: the cpu to read the buffer from 5918 * @ts: a variable to store the timestamp (may be NULL) 5919 * @lost_events: a variable to store if events were lost (may be NULL) 5920 * 5921 * Returns the next event in the ring buffer, and that event is consumed. 5922 * Meaning, that sequential reads will keep returning a different event, 5923 * and eventually empty the ring buffer if the producer is slower. 5924 */ 5925struct ring_buffer_event * 5926ring_buffer_consume(struct trace_buffer *buffer, int cpu, u64 *ts, 5927 unsigned long *lost_events) 5928{ 5929 struct ring_buffer_per_cpu *cpu_buffer; 5930 struct ring_buffer_event *event = NULL; 5931 unsigned long flags; 5932 bool dolock; 5933 5934 again: 5935 /* might be called in atomic */ 5936 preempt_disable(); 5937 5938 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 5939 goto out; 5940 5941 cpu_buffer = buffer->buffers[cpu]; 5942 local_irq_save(flags); 5943 dolock = rb_reader_lock(cpu_buffer); 5944 5945 event = rb_buffer_peek(cpu_buffer, ts, lost_events); 5946 if (event) { 5947 cpu_buffer->lost_events = 0; 5948 rb_advance_reader(cpu_buffer); 5949 } 5950 5951 rb_reader_unlock(cpu_buffer, dolock); 5952 local_irq_restore(flags); 5953 5954 out: 5955 preempt_enable(); 5956 5957 if (event && event->type_len == RINGBUF_TYPE_PADDING) 5958 goto again; 5959 5960 return event; 5961} 5962EXPORT_SYMBOL_GPL(ring_buffer_consume); 5963 5964/** 5965 * ring_buffer_read_start - start a non consuming read of the buffer 5966 * @buffer: The ring buffer to read from 5967 * @cpu: The cpu buffer to iterate over 5968 * @flags: gfp flags to use for memory allocation 5969 * 5970 * This creates an iterator to allow non-consuming iteration through 5971 * the buffer. If the buffer is disabled for writing, it will produce 5972 * the same information each time, but if the buffer is still writing 5973 * then the first hit of a write will cause the iteration to stop. 5974 * 5975 * Must be paired with ring_buffer_read_finish. 5976 */ 5977struct ring_buffer_iter * 5978ring_buffer_read_start(struct trace_buffer *buffer, int cpu, gfp_t flags) 5979{ 5980 struct ring_buffer_per_cpu *cpu_buffer; 5981 struct ring_buffer_iter *iter; 5982 5983 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 5984 return NULL; 5985 5986 iter = kzalloc(sizeof(*iter), flags); 5987 if (!iter) 5988 return NULL; 5989 5990 /* Holds the entire event: data and meta data */ 5991 iter->event_size = buffer->subbuf_size; 5992 iter->event = kmalloc(iter->event_size, flags); 5993 if (!iter->event) { 5994 kfree(iter); 5995 return NULL; 5996 } 5997 5998 cpu_buffer = buffer->buffers[cpu]; 5999 6000 iter->cpu_buffer = cpu_buffer; 6001 6002 atomic_inc(&cpu_buffer->resize_disabled); 6003 6004 guard(raw_spinlock_irqsave)(&cpu_buffer->reader_lock); 6005 arch_spin_lock(&cpu_buffer->lock); 6006 rb_iter_reset(iter); 6007 arch_spin_unlock(&cpu_buffer->lock); 6008 6009 return iter; 6010} 6011EXPORT_SYMBOL_GPL(ring_buffer_read_start); 6012 6013/** 6014 * ring_buffer_read_finish - finish reading the iterator of the buffer 6015 * @iter: The iterator retrieved by ring_buffer_start 6016 * 6017 * This re-enables resizing of the buffer, and frees the iterator. 6018 */ 6019void 6020ring_buffer_read_finish(struct ring_buffer_iter *iter) 6021{ 6022 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer; 6023 6024 /* Use this opportunity to check the integrity of the ring buffer. */ 6025 rb_check_pages(cpu_buffer); 6026 6027 atomic_dec(&cpu_buffer->resize_disabled); 6028 kfree(iter->event); 6029 kfree(iter); 6030} 6031EXPORT_SYMBOL_GPL(ring_buffer_read_finish); 6032 6033/** 6034 * ring_buffer_iter_advance - advance the iterator to the next location 6035 * @iter: The ring buffer iterator 6036 * 6037 * Move the location of the iterator such that the next read will 6038 * be the next location of the iterator. 6039 */ 6040void ring_buffer_iter_advance(struct ring_buffer_iter *iter) 6041{ 6042 struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer; 6043 unsigned long flags; 6044 6045 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 6046 6047 rb_advance_iter(iter); 6048 6049 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 6050} 6051EXPORT_SYMBOL_GPL(ring_buffer_iter_advance); 6052 6053/** 6054 * ring_buffer_size - return the size of the ring buffer (in bytes) 6055 * @buffer: The ring buffer. 6056 * @cpu: The CPU to get ring buffer size from. 6057 */ 6058unsigned long ring_buffer_size(struct trace_buffer *buffer, int cpu) 6059{ 6060 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 6061 return 0; 6062 6063 return buffer->subbuf_size * buffer->buffers[cpu]->nr_pages; 6064} 6065EXPORT_SYMBOL_GPL(ring_buffer_size); 6066 6067/** 6068 * ring_buffer_max_event_size - return the max data size of an event 6069 * @buffer: The ring buffer. 6070 * 6071 * Returns the maximum size an event can be. 6072 */ 6073unsigned long ring_buffer_max_event_size(struct trace_buffer *buffer) 6074{ 6075 /* If abs timestamp is requested, events have a timestamp too */ 6076 if (ring_buffer_time_stamp_abs(buffer)) 6077 return buffer->max_data_size - RB_LEN_TIME_EXTEND; 6078 return buffer->max_data_size; 6079} 6080EXPORT_SYMBOL_GPL(ring_buffer_max_event_size); 6081 6082static void rb_clear_buffer_page(struct buffer_page *page) 6083{ 6084 local_set(&page->write, 0); 6085 local_set(&page->entries, 0); 6086 rb_init_page(page->page); 6087 page->read = 0; 6088} 6089 6090/* 6091 * When the buffer is memory mapped to user space, each sub buffer 6092 * has a unique id that is used by the meta data to tell the user 6093 * where the current reader page is. 6094 * 6095 * For a normal allocated ring buffer, the id is saved in the buffer page 6096 * id field, and updated via this function. 6097 * 6098 * But for a fixed memory mapped buffer, the id is already assigned for 6099 * fixed memory ordering in the memory layout and can not be used. Instead 6100 * the index of where the page lies in the memory layout is used. 6101 * 6102 * For the normal pages, set the buffer page id with the passed in @id 6103 * value and return that. 6104 * 6105 * For fixed memory mapped pages, get the page index in the memory layout 6106 * and return that as the id. 6107 */ 6108static int rb_page_id(struct ring_buffer_per_cpu *cpu_buffer, 6109 struct buffer_page *bpage, int id) 6110{ 6111 /* 6112 * For boot buffers, the id is the index, 6113 * otherwise, set the buffer page with this id 6114 */ 6115 if (cpu_buffer->ring_meta) 6116 id = rb_meta_subbuf_idx(cpu_buffer->ring_meta, bpage->page); 6117 else 6118 bpage->id = id; 6119 6120 return id; 6121} 6122 6123static void rb_update_meta_page(struct ring_buffer_per_cpu *cpu_buffer) 6124{ 6125 struct trace_buffer_meta *meta = cpu_buffer->meta_page; 6126 6127 if (!meta) 6128 return; 6129 6130 meta->reader.read = cpu_buffer->reader_page->read; 6131 meta->reader.id = rb_page_id(cpu_buffer, cpu_buffer->reader_page, 6132 cpu_buffer->reader_page->id); 6133 6134 meta->reader.lost_events = cpu_buffer->lost_events; 6135 6136 meta->entries = local_read(&cpu_buffer->entries); 6137 meta->overrun = local_read(&cpu_buffer->overrun); 6138 meta->read = cpu_buffer->read; 6139 6140 /* Some archs do not have data cache coherency between kernel and user-space */ 6141 flush_kernel_vmap_range(cpu_buffer->meta_page, PAGE_SIZE); 6142} 6143 6144static void 6145rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer) 6146{ 6147 struct buffer_page *page; 6148 6149 rb_head_page_deactivate(cpu_buffer); 6150 6151 cpu_buffer->head_page 6152 = list_entry(cpu_buffer->pages, struct buffer_page, list); 6153 rb_clear_buffer_page(cpu_buffer->head_page); 6154 list_for_each_entry(page, cpu_buffer->pages, list) { 6155 rb_clear_buffer_page(page); 6156 } 6157 6158 cpu_buffer->tail_page = cpu_buffer->head_page; 6159 cpu_buffer->commit_page = cpu_buffer->head_page; 6160 6161 INIT_LIST_HEAD(&cpu_buffer->reader_page->list); 6162 INIT_LIST_HEAD(&cpu_buffer->new_pages); 6163 rb_clear_buffer_page(cpu_buffer->reader_page); 6164 6165 local_set(&cpu_buffer->entries_bytes, 0); 6166 local_set(&cpu_buffer->overrun, 0); 6167 local_set(&cpu_buffer->commit_overrun, 0); 6168 local_set(&cpu_buffer->dropped_events, 0); 6169 local_set(&cpu_buffer->entries, 0); 6170 local_set(&cpu_buffer->committing, 0); 6171 local_set(&cpu_buffer->commits, 0); 6172 local_set(&cpu_buffer->pages_touched, 0); 6173 local_set(&cpu_buffer->pages_lost, 0); 6174 local_set(&cpu_buffer->pages_read, 0); 6175 cpu_buffer->last_pages_touch = 0; 6176 cpu_buffer->shortest_full = 0; 6177 cpu_buffer->read = 0; 6178 cpu_buffer->read_bytes = 0; 6179 6180 rb_time_set(&cpu_buffer->write_stamp, 0); 6181 rb_time_set(&cpu_buffer->before_stamp, 0); 6182 6183 memset(cpu_buffer->event_stamp, 0, sizeof(cpu_buffer->event_stamp)); 6184 6185 cpu_buffer->lost_events = 0; 6186 cpu_buffer->last_overrun = 0; 6187 6188 rb_head_page_activate(cpu_buffer); 6189 cpu_buffer->pages_removed = 0; 6190 6191 if (cpu_buffer->mapped) { 6192 rb_update_meta_page(cpu_buffer); 6193 if (cpu_buffer->ring_meta) { 6194 struct ring_buffer_cpu_meta *meta = cpu_buffer->ring_meta; 6195 meta->commit_buffer = meta->head_buffer; 6196 } 6197 } 6198} 6199 6200/* Must have disabled the cpu buffer then done a synchronize_rcu */ 6201static void reset_disabled_cpu_buffer(struct ring_buffer_per_cpu *cpu_buffer) 6202{ 6203 guard(raw_spinlock_irqsave)(&cpu_buffer->reader_lock); 6204 6205 if (RB_WARN_ON(cpu_buffer, local_read(&cpu_buffer->committing))) 6206 return; 6207 6208 arch_spin_lock(&cpu_buffer->lock); 6209 6210 rb_reset_cpu(cpu_buffer); 6211 6212 arch_spin_unlock(&cpu_buffer->lock); 6213} 6214 6215/** 6216 * ring_buffer_reset_cpu - reset a ring buffer per CPU buffer 6217 * @buffer: The ring buffer to reset a per cpu buffer of 6218 * @cpu: The CPU buffer to be reset 6219 */ 6220void ring_buffer_reset_cpu(struct trace_buffer *buffer, int cpu) 6221{ 6222 struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu]; 6223 6224 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 6225 return; 6226 6227 /* prevent another thread from changing buffer sizes */ 6228 mutex_lock(&buffer->mutex); 6229 6230 atomic_inc(&cpu_buffer->resize_disabled); 6231 atomic_inc(&cpu_buffer->record_disabled); 6232 6233 /* Make sure all commits have finished */ 6234 synchronize_rcu(); 6235 6236 reset_disabled_cpu_buffer(cpu_buffer); 6237 6238 atomic_dec(&cpu_buffer->record_disabled); 6239 atomic_dec(&cpu_buffer->resize_disabled); 6240 6241 mutex_unlock(&buffer->mutex); 6242} 6243EXPORT_SYMBOL_GPL(ring_buffer_reset_cpu); 6244 6245/* Flag to ensure proper resetting of atomic variables */ 6246#define RESET_BIT (1 << 30) 6247 6248/** 6249 * ring_buffer_reset_online_cpus - reset a ring buffer per CPU buffer 6250 * @buffer: The ring buffer to reset a per cpu buffer of 6251 */ 6252void ring_buffer_reset_online_cpus(struct trace_buffer *buffer) 6253{ 6254 struct ring_buffer_per_cpu *cpu_buffer; 6255 int cpu; 6256 6257 /* prevent another thread from changing buffer sizes */ 6258 mutex_lock(&buffer->mutex); 6259 6260 for_each_online_buffer_cpu(buffer, cpu) { 6261 cpu_buffer = buffer->buffers[cpu]; 6262 6263 atomic_add(RESET_BIT, &cpu_buffer->resize_disabled); 6264 atomic_inc(&cpu_buffer->record_disabled); 6265 } 6266 6267 /* Make sure all commits have finished */ 6268 synchronize_rcu(); 6269 6270 for_each_buffer_cpu(buffer, cpu) { 6271 cpu_buffer = buffer->buffers[cpu]; 6272 6273 /* 6274 * If a CPU came online during the synchronize_rcu(), then 6275 * ignore it. 6276 */ 6277 if (!(atomic_read(&cpu_buffer->resize_disabled) & RESET_BIT)) 6278 continue; 6279 6280 reset_disabled_cpu_buffer(cpu_buffer); 6281 6282 atomic_dec(&cpu_buffer->record_disabled); 6283 atomic_sub(RESET_BIT, &cpu_buffer->resize_disabled); 6284 } 6285 6286 mutex_unlock(&buffer->mutex); 6287} 6288 6289/** 6290 * ring_buffer_reset - reset a ring buffer 6291 * @buffer: The ring buffer to reset all cpu buffers 6292 */ 6293void ring_buffer_reset(struct trace_buffer *buffer) 6294{ 6295 struct ring_buffer_per_cpu *cpu_buffer; 6296 int cpu; 6297 6298 /* prevent another thread from changing buffer sizes */ 6299 mutex_lock(&buffer->mutex); 6300 6301 for_each_buffer_cpu(buffer, cpu) { 6302 cpu_buffer = buffer->buffers[cpu]; 6303 6304 atomic_inc(&cpu_buffer->resize_disabled); 6305 atomic_inc(&cpu_buffer->record_disabled); 6306 } 6307 6308 /* Make sure all commits have finished */ 6309 synchronize_rcu(); 6310 6311 for_each_buffer_cpu(buffer, cpu) { 6312 cpu_buffer = buffer->buffers[cpu]; 6313 6314 reset_disabled_cpu_buffer(cpu_buffer); 6315 6316 atomic_dec(&cpu_buffer->record_disabled); 6317 atomic_dec(&cpu_buffer->resize_disabled); 6318 } 6319 6320 mutex_unlock(&buffer->mutex); 6321} 6322EXPORT_SYMBOL_GPL(ring_buffer_reset); 6323 6324/** 6325 * ring_buffer_empty - is the ring buffer empty? 6326 * @buffer: The ring buffer to test 6327 */ 6328bool ring_buffer_empty(struct trace_buffer *buffer) 6329{ 6330 struct ring_buffer_per_cpu *cpu_buffer; 6331 unsigned long flags; 6332 bool dolock; 6333 bool ret; 6334 int cpu; 6335 6336 /* yes this is racy, but if you don't like the race, lock the buffer */ 6337 for_each_buffer_cpu(buffer, cpu) { 6338 cpu_buffer = buffer->buffers[cpu]; 6339 local_irq_save(flags); 6340 dolock = rb_reader_lock(cpu_buffer); 6341 ret = rb_per_cpu_empty(cpu_buffer); 6342 rb_reader_unlock(cpu_buffer, dolock); 6343 local_irq_restore(flags); 6344 6345 if (!ret) 6346 return false; 6347 } 6348 6349 return true; 6350} 6351EXPORT_SYMBOL_GPL(ring_buffer_empty); 6352 6353/** 6354 * ring_buffer_empty_cpu - is a cpu buffer of a ring buffer empty? 6355 * @buffer: The ring buffer 6356 * @cpu: The CPU buffer to test 6357 */ 6358bool ring_buffer_empty_cpu(struct trace_buffer *buffer, int cpu) 6359{ 6360 struct ring_buffer_per_cpu *cpu_buffer; 6361 unsigned long flags; 6362 bool dolock; 6363 bool ret; 6364 6365 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 6366 return true; 6367 6368 cpu_buffer = buffer->buffers[cpu]; 6369 local_irq_save(flags); 6370 dolock = rb_reader_lock(cpu_buffer); 6371 ret = rb_per_cpu_empty(cpu_buffer); 6372 rb_reader_unlock(cpu_buffer, dolock); 6373 local_irq_restore(flags); 6374 6375 return ret; 6376} 6377EXPORT_SYMBOL_GPL(ring_buffer_empty_cpu); 6378 6379#ifdef CONFIG_RING_BUFFER_ALLOW_SWAP 6380/** 6381 * ring_buffer_swap_cpu - swap a CPU buffer between two ring buffers 6382 * @buffer_a: One buffer to swap with 6383 * @buffer_b: The other buffer to swap with 6384 * @cpu: the CPU of the buffers to swap 6385 * 6386 * This function is useful for tracers that want to take a "snapshot" 6387 * of a CPU buffer and has another back up buffer lying around. 6388 * it is expected that the tracer handles the cpu buffer not being 6389 * used at the moment. 6390 */ 6391int ring_buffer_swap_cpu(struct trace_buffer *buffer_a, 6392 struct trace_buffer *buffer_b, int cpu) 6393{ 6394 struct ring_buffer_per_cpu *cpu_buffer_a; 6395 struct ring_buffer_per_cpu *cpu_buffer_b; 6396 int ret = -EINVAL; 6397 6398 if (!cpumask_test_cpu(cpu, buffer_a->cpumask) || 6399 !cpumask_test_cpu(cpu, buffer_b->cpumask)) 6400 return -EINVAL; 6401 6402 cpu_buffer_a = buffer_a->buffers[cpu]; 6403 cpu_buffer_b = buffer_b->buffers[cpu]; 6404 6405 /* It's up to the callers to not try to swap mapped buffers */ 6406 if (WARN_ON_ONCE(cpu_buffer_a->mapped || cpu_buffer_b->mapped)) 6407 return -EBUSY; 6408 6409 /* At least make sure the two buffers are somewhat the same */ 6410 if (cpu_buffer_a->nr_pages != cpu_buffer_b->nr_pages) 6411 return -EINVAL; 6412 6413 if (buffer_a->subbuf_order != buffer_b->subbuf_order) 6414 return -EINVAL; 6415 6416 if (atomic_read(&buffer_a->record_disabled)) 6417 return -EAGAIN; 6418 6419 if (atomic_read(&buffer_b->record_disabled)) 6420 return -EAGAIN; 6421 6422 if (atomic_read(&cpu_buffer_a->record_disabled)) 6423 return -EAGAIN; 6424 6425 if (atomic_read(&cpu_buffer_b->record_disabled)) 6426 return -EAGAIN; 6427 6428 /* 6429 * We can't do a synchronize_rcu here because this 6430 * function can be called in atomic context. 6431 * Normally this will be called from the same CPU as cpu. 6432 * If not it's up to the caller to protect this. 6433 */ 6434 atomic_inc(&cpu_buffer_a->record_disabled); 6435 atomic_inc(&cpu_buffer_b->record_disabled); 6436 6437 ret = -EBUSY; 6438 if (local_read(&cpu_buffer_a->committing)) 6439 goto out_dec; 6440 if (local_read(&cpu_buffer_b->committing)) 6441 goto out_dec; 6442 6443 /* 6444 * When resize is in progress, we cannot swap it because 6445 * it will mess the state of the cpu buffer. 6446 */ 6447 if (atomic_read(&buffer_a->resizing)) 6448 goto out_dec; 6449 if (atomic_read(&buffer_b->resizing)) 6450 goto out_dec; 6451 6452 buffer_a->buffers[cpu] = cpu_buffer_b; 6453 buffer_b->buffers[cpu] = cpu_buffer_a; 6454 6455 cpu_buffer_b->buffer = buffer_a; 6456 cpu_buffer_a->buffer = buffer_b; 6457 6458 ret = 0; 6459 6460out_dec: 6461 atomic_dec(&cpu_buffer_a->record_disabled); 6462 atomic_dec(&cpu_buffer_b->record_disabled); 6463 return ret; 6464} 6465EXPORT_SYMBOL_GPL(ring_buffer_swap_cpu); 6466#endif /* CONFIG_RING_BUFFER_ALLOW_SWAP */ 6467 6468/** 6469 * ring_buffer_alloc_read_page - allocate a page to read from buffer 6470 * @buffer: the buffer to allocate for. 6471 * @cpu: the cpu buffer to allocate. 6472 * 6473 * This function is used in conjunction with ring_buffer_read_page. 6474 * When reading a full page from the ring buffer, these functions 6475 * can be used to speed up the process. The calling function should 6476 * allocate a few pages first with this function. Then when it 6477 * needs to get pages from the ring buffer, it passes the result 6478 * of this function into ring_buffer_read_page, which will swap 6479 * the page that was allocated, with the read page of the buffer. 6480 * 6481 * Returns: 6482 * The page allocated, or ERR_PTR 6483 */ 6484struct buffer_data_read_page * 6485ring_buffer_alloc_read_page(struct trace_buffer *buffer, int cpu) 6486{ 6487 struct ring_buffer_per_cpu *cpu_buffer; 6488 struct buffer_data_read_page *bpage = NULL; 6489 unsigned long flags; 6490 6491 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 6492 return ERR_PTR(-ENODEV); 6493 6494 bpage = kzalloc(sizeof(*bpage), GFP_KERNEL); 6495 if (!bpage) 6496 return ERR_PTR(-ENOMEM); 6497 6498 bpage->order = buffer->subbuf_order; 6499 cpu_buffer = buffer->buffers[cpu]; 6500 local_irq_save(flags); 6501 arch_spin_lock(&cpu_buffer->lock); 6502 6503 if (cpu_buffer->free_page) { 6504 bpage->data = cpu_buffer->free_page; 6505 cpu_buffer->free_page = NULL; 6506 } 6507 6508 arch_spin_unlock(&cpu_buffer->lock); 6509 local_irq_restore(flags); 6510 6511 if (bpage->data) { 6512 rb_init_page(bpage->data); 6513 } else { 6514 bpage->data = alloc_cpu_data(cpu, cpu_buffer->buffer->subbuf_order); 6515 if (!bpage->data) { 6516 kfree(bpage); 6517 return ERR_PTR(-ENOMEM); 6518 } 6519 } 6520 6521 return bpage; 6522} 6523EXPORT_SYMBOL_GPL(ring_buffer_alloc_read_page); 6524 6525/** 6526 * ring_buffer_free_read_page - free an allocated read page 6527 * @buffer: the buffer the page was allocate for 6528 * @cpu: the cpu buffer the page came from 6529 * @data_page: the page to free 6530 * 6531 * Free a page allocated from ring_buffer_alloc_read_page. 6532 */ 6533void ring_buffer_free_read_page(struct trace_buffer *buffer, int cpu, 6534 struct buffer_data_read_page *data_page) 6535{ 6536 struct ring_buffer_per_cpu *cpu_buffer; 6537 struct buffer_data_page *bpage = data_page->data; 6538 struct page *page = virt_to_page(bpage); 6539 unsigned long flags; 6540 6541 if (!buffer || !buffer->buffers || !buffer->buffers[cpu]) 6542 return; 6543 6544 cpu_buffer = buffer->buffers[cpu]; 6545 6546 /* 6547 * If the page is still in use someplace else, or order of the page 6548 * is different from the subbuffer order of the buffer - 6549 * we can't reuse it 6550 */ 6551 if (page_ref_count(page) > 1 || data_page->order != buffer->subbuf_order) 6552 goto out; 6553 6554 local_irq_save(flags); 6555 arch_spin_lock(&cpu_buffer->lock); 6556 6557 if (!cpu_buffer->free_page) { 6558 cpu_buffer->free_page = bpage; 6559 bpage = NULL; 6560 } 6561 6562 arch_spin_unlock(&cpu_buffer->lock); 6563 local_irq_restore(flags); 6564 6565 out: 6566 free_pages((unsigned long)bpage, data_page->order); 6567 kfree(data_page); 6568} 6569EXPORT_SYMBOL_GPL(ring_buffer_free_read_page); 6570 6571/** 6572 * ring_buffer_read_page - extract a page from the ring buffer 6573 * @buffer: buffer to extract from 6574 * @data_page: the page to use allocated from ring_buffer_alloc_read_page 6575 * @len: amount to extract 6576 * @cpu: the cpu of the buffer to extract 6577 * @full: should the extraction only happen when the page is full. 6578 * 6579 * This function will pull out a page from the ring buffer and consume it. 6580 * @data_page must be the address of the variable that was returned 6581 * from ring_buffer_alloc_read_page. This is because the page might be used 6582 * to swap with a page in the ring buffer. 6583 * 6584 * for example: 6585 * rpage = ring_buffer_alloc_read_page(buffer, cpu); 6586 * if (IS_ERR(rpage)) 6587 * return PTR_ERR(rpage); 6588 * ret = ring_buffer_read_page(buffer, rpage, len, cpu, 0); 6589 * if (ret >= 0) 6590 * process_page(ring_buffer_read_page_data(rpage), ret); 6591 * ring_buffer_free_read_page(buffer, cpu, rpage); 6592 * 6593 * When @full is set, the function will not return true unless 6594 * the writer is off the reader page. 6595 * 6596 * Note: it is up to the calling functions to handle sleeps and wakeups. 6597 * The ring buffer can be used anywhere in the kernel and can not 6598 * blindly call wake_up. The layer that uses the ring buffer must be 6599 * responsible for that. 6600 * 6601 * Returns: 6602 * >=0 if data has been transferred, returns the offset of consumed data. 6603 * <0 if no data has been transferred. 6604 */ 6605int ring_buffer_read_page(struct trace_buffer *buffer, 6606 struct buffer_data_read_page *data_page, 6607 size_t len, int cpu, int full) 6608{ 6609 struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu]; 6610 struct ring_buffer_event *event; 6611 struct buffer_data_page *bpage; 6612 struct buffer_page *reader; 6613 unsigned long missed_events; 6614 unsigned int commit; 6615 unsigned int read; 6616 u64 save_timestamp; 6617 6618 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 6619 return -1; 6620 6621 /* 6622 * If len is not big enough to hold the page header, then 6623 * we can not copy anything. 6624 */ 6625 if (len <= BUF_PAGE_HDR_SIZE) 6626 return -1; 6627 6628 len -= BUF_PAGE_HDR_SIZE; 6629 6630 if (!data_page || !data_page->data) 6631 return -1; 6632 6633 if (data_page->order != buffer->subbuf_order) 6634 return -1; 6635 6636 bpage = data_page->data; 6637 if (!bpage) 6638 return -1; 6639 6640 guard(raw_spinlock_irqsave)(&cpu_buffer->reader_lock); 6641 6642 reader = rb_get_reader_page(cpu_buffer); 6643 if (!reader) 6644 return -1; 6645 6646 event = rb_reader_event(cpu_buffer); 6647 6648 read = reader->read; 6649 commit = rb_page_size(reader); 6650 6651 /* Check if any events were dropped */ 6652 missed_events = cpu_buffer->lost_events; 6653 6654 /* 6655 * If this page has been partially read or 6656 * if len is not big enough to read the rest of the page or 6657 * a writer is still on the page, then 6658 * we must copy the data from the page to the buffer. 6659 * Otherwise, we can simply swap the page with the one passed in. 6660 */ 6661 if (read || (len < (commit - read)) || 6662 cpu_buffer->reader_page == cpu_buffer->commit_page || 6663 cpu_buffer->mapped) { 6664 struct buffer_data_page *rpage = cpu_buffer->reader_page->page; 6665 unsigned int rpos = read; 6666 unsigned int pos = 0; 6667 unsigned int size; 6668 6669 /* 6670 * If a full page is expected, this can still be returned 6671 * if there's been a previous partial read and the 6672 * rest of the page can be read and the commit page is off 6673 * the reader page. 6674 */ 6675 if (full && 6676 (!read || (len < (commit - read)) || 6677 cpu_buffer->reader_page == cpu_buffer->commit_page)) 6678 return -1; 6679 6680 if (len > (commit - read)) 6681 len = (commit - read); 6682 6683 /* Always keep the time extend and data together */ 6684 size = rb_event_ts_length(event); 6685 6686 if (len < size) 6687 return -1; 6688 6689 /* save the current timestamp, since the user will need it */ 6690 save_timestamp = cpu_buffer->read_stamp; 6691 6692 /* Need to copy one event at a time */ 6693 do { 6694 /* We need the size of one event, because 6695 * rb_advance_reader only advances by one event, 6696 * whereas rb_event_ts_length may include the size of 6697 * one or two events. 6698 * We have already ensured there's enough space if this 6699 * is a time extend. */ 6700 size = rb_event_length(event); 6701 memcpy(bpage->data + pos, rpage->data + rpos, size); 6702 6703 len -= size; 6704 6705 rb_advance_reader(cpu_buffer); 6706 rpos = reader->read; 6707 pos += size; 6708 6709 if (rpos >= commit) 6710 break; 6711 6712 event = rb_reader_event(cpu_buffer); 6713 /* Always keep the time extend and data together */ 6714 size = rb_event_ts_length(event); 6715 } while (len >= size); 6716 6717 /* update bpage */ 6718 local_set(&bpage->commit, pos); 6719 bpage->time_stamp = save_timestamp; 6720 6721 /* we copied everything to the beginning */ 6722 read = 0; 6723 } else { 6724 /* update the entry counter */ 6725 cpu_buffer->read += rb_page_entries(reader); 6726 cpu_buffer->read_bytes += rb_page_size(reader); 6727 6728 /* swap the pages */ 6729 rb_init_page(bpage); 6730 bpage = reader->page; 6731 reader->page = data_page->data; 6732 local_set(&reader->write, 0); 6733 local_set(&reader->entries, 0); 6734 reader->read = 0; 6735 data_page->data = bpage; 6736 6737 /* 6738 * Use the real_end for the data size, 6739 * This gives us a chance to store the lost events 6740 * on the page. 6741 */ 6742 if (reader->real_end) 6743 local_set(&bpage->commit, reader->real_end); 6744 } 6745 6746 cpu_buffer->lost_events = 0; 6747 6748 commit = local_read(&bpage->commit); 6749 /* 6750 * Set a flag in the commit field if we lost events 6751 */ 6752 if (missed_events) { 6753 /* If there is room at the end of the page to save the 6754 * missed events, then record it there. 6755 */ 6756 if (buffer->subbuf_size - commit >= sizeof(missed_events)) { 6757 memcpy(&bpage->data[commit], &missed_events, 6758 sizeof(missed_events)); 6759 local_add(RB_MISSED_STORED, &bpage->commit); 6760 commit += sizeof(missed_events); 6761 } 6762 local_add(RB_MISSED_EVENTS, &bpage->commit); 6763 } 6764 6765 /* 6766 * This page may be off to user land. Zero it out here. 6767 */ 6768 if (commit < buffer->subbuf_size) 6769 memset(&bpage->data[commit], 0, buffer->subbuf_size - commit); 6770 6771 return read; 6772} 6773EXPORT_SYMBOL_GPL(ring_buffer_read_page); 6774 6775/** 6776 * ring_buffer_read_page_data - get pointer to the data in the page. 6777 * @page: the page to get the data from 6778 * 6779 * Returns pointer to the actual data in this page. 6780 */ 6781void *ring_buffer_read_page_data(struct buffer_data_read_page *page) 6782{ 6783 return page->data; 6784} 6785EXPORT_SYMBOL_GPL(ring_buffer_read_page_data); 6786 6787/** 6788 * ring_buffer_subbuf_size_get - get size of the sub buffer. 6789 * @buffer: the buffer to get the sub buffer size from 6790 * 6791 * Returns size of the sub buffer, in bytes. 6792 */ 6793int ring_buffer_subbuf_size_get(struct trace_buffer *buffer) 6794{ 6795 return buffer->subbuf_size + BUF_PAGE_HDR_SIZE; 6796} 6797EXPORT_SYMBOL_GPL(ring_buffer_subbuf_size_get); 6798 6799/** 6800 * ring_buffer_subbuf_order_get - get order of system sub pages in one buffer page. 6801 * @buffer: The ring_buffer to get the system sub page order from 6802 * 6803 * By default, one ring buffer sub page equals to one system page. This parameter 6804 * is configurable, per ring buffer. The size of the ring buffer sub page can be 6805 * extended, but must be an order of system page size. 6806 * 6807 * Returns the order of buffer sub page size, in system pages: 6808 * 0 means the sub buffer size is 1 system page and so forth. 6809 * In case of an error < 0 is returned. 6810 */ 6811int ring_buffer_subbuf_order_get(struct trace_buffer *buffer) 6812{ 6813 if (!buffer) 6814 return -EINVAL; 6815 6816 return buffer->subbuf_order; 6817} 6818EXPORT_SYMBOL_GPL(ring_buffer_subbuf_order_get); 6819 6820/** 6821 * ring_buffer_subbuf_order_set - set the size of ring buffer sub page. 6822 * @buffer: The ring_buffer to set the new page size. 6823 * @order: Order of the system pages in one sub buffer page 6824 * 6825 * By default, one ring buffer pages equals to one system page. This API can be 6826 * used to set new size of the ring buffer page. The size must be order of 6827 * system page size, that's why the input parameter @order is the order of 6828 * system pages that are allocated for one ring buffer page: 6829 * 0 - 1 system page 6830 * 1 - 2 system pages 6831 * 3 - 4 system pages 6832 * ... 6833 * 6834 * Returns 0 on success or < 0 in case of an error. 6835 */ 6836int ring_buffer_subbuf_order_set(struct trace_buffer *buffer, int order) 6837{ 6838 struct ring_buffer_per_cpu *cpu_buffer; 6839 struct buffer_page *bpage, *tmp; 6840 int old_order, old_size; 6841 int nr_pages; 6842 int psize; 6843 int err; 6844 int cpu; 6845 6846 if (!buffer || order < 0) 6847 return -EINVAL; 6848 6849 if (buffer->subbuf_order == order) 6850 return 0; 6851 6852 psize = (1 << order) * PAGE_SIZE; 6853 if (psize <= BUF_PAGE_HDR_SIZE) 6854 return -EINVAL; 6855 6856 /* Size of a subbuf cannot be greater than the write counter */ 6857 if (psize > RB_WRITE_MASK + 1) 6858 return -EINVAL; 6859 6860 old_order = buffer->subbuf_order; 6861 old_size = buffer->subbuf_size; 6862 6863 /* prevent another thread from changing buffer sizes */ 6864 guard(mutex)(&buffer->mutex); 6865 atomic_inc(&buffer->record_disabled); 6866 6867 /* Make sure all commits have finished */ 6868 synchronize_rcu(); 6869 6870 buffer->subbuf_order = order; 6871 buffer->subbuf_size = psize - BUF_PAGE_HDR_SIZE; 6872 6873 /* Make sure all new buffers are allocated, before deleting the old ones */ 6874 for_each_buffer_cpu(buffer, cpu) { 6875 6876 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 6877 continue; 6878 6879 cpu_buffer = buffer->buffers[cpu]; 6880 6881 if (cpu_buffer->mapped) { 6882 err = -EBUSY; 6883 goto error; 6884 } 6885 6886 /* Update the number of pages to match the new size */ 6887 nr_pages = old_size * buffer->buffers[cpu]->nr_pages; 6888 nr_pages = DIV_ROUND_UP(nr_pages, buffer->subbuf_size); 6889 6890 /* we need a minimum of two pages */ 6891 if (nr_pages < 2) 6892 nr_pages = 2; 6893 6894 cpu_buffer->nr_pages_to_update = nr_pages; 6895 6896 /* Include the reader page */ 6897 nr_pages++; 6898 6899 /* Allocate the new size buffer */ 6900 INIT_LIST_HEAD(&cpu_buffer->new_pages); 6901 if (__rb_allocate_pages(cpu_buffer, nr_pages, 6902 &cpu_buffer->new_pages)) { 6903 /* not enough memory for new pages */ 6904 err = -ENOMEM; 6905 goto error; 6906 } 6907 } 6908 6909 for_each_buffer_cpu(buffer, cpu) { 6910 struct buffer_data_page *old_free_data_page; 6911 struct list_head old_pages; 6912 unsigned long flags; 6913 6914 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 6915 continue; 6916 6917 cpu_buffer = buffer->buffers[cpu]; 6918 6919 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 6920 6921 /* Clear the head bit to make the link list normal to read */ 6922 rb_head_page_deactivate(cpu_buffer); 6923 6924 /* 6925 * Collect buffers from the cpu_buffer pages list and the 6926 * reader_page on old_pages, so they can be freed later when not 6927 * under a spinlock. The pages list is a linked list with no 6928 * head, adding old_pages turns it into a regular list with 6929 * old_pages being the head. 6930 */ 6931 list_add(&old_pages, cpu_buffer->pages); 6932 list_add(&cpu_buffer->reader_page->list, &old_pages); 6933 6934 /* One page was allocated for the reader page */ 6935 cpu_buffer->reader_page = list_entry(cpu_buffer->new_pages.next, 6936 struct buffer_page, list); 6937 list_del_init(&cpu_buffer->reader_page->list); 6938 6939 /* Install the new pages, remove the head from the list */ 6940 cpu_buffer->pages = cpu_buffer->new_pages.next; 6941 list_del_init(&cpu_buffer->new_pages); 6942 cpu_buffer->cnt++; 6943 6944 cpu_buffer->head_page 6945 = list_entry(cpu_buffer->pages, struct buffer_page, list); 6946 cpu_buffer->tail_page = cpu_buffer->commit_page = cpu_buffer->head_page; 6947 6948 cpu_buffer->nr_pages = cpu_buffer->nr_pages_to_update; 6949 cpu_buffer->nr_pages_to_update = 0; 6950 6951 old_free_data_page = cpu_buffer->free_page; 6952 cpu_buffer->free_page = NULL; 6953 6954 rb_head_page_activate(cpu_buffer); 6955 6956 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 6957 6958 /* Free old sub buffers */ 6959 list_for_each_entry_safe(bpage, tmp, &old_pages, list) { 6960 list_del_init(&bpage->list); 6961 free_buffer_page(bpage); 6962 } 6963 free_pages((unsigned long)old_free_data_page, old_order); 6964 6965 rb_check_pages(cpu_buffer); 6966 } 6967 6968 atomic_dec(&buffer->record_disabled); 6969 6970 return 0; 6971 6972error: 6973 buffer->subbuf_order = old_order; 6974 buffer->subbuf_size = old_size; 6975 6976 atomic_dec(&buffer->record_disabled); 6977 6978 for_each_buffer_cpu(buffer, cpu) { 6979 cpu_buffer = buffer->buffers[cpu]; 6980 6981 if (!cpu_buffer->nr_pages_to_update) 6982 continue; 6983 6984 list_for_each_entry_safe(bpage, tmp, &cpu_buffer->new_pages, list) { 6985 list_del_init(&bpage->list); 6986 free_buffer_page(bpage); 6987 } 6988 } 6989 6990 return err; 6991} 6992EXPORT_SYMBOL_GPL(ring_buffer_subbuf_order_set); 6993 6994static int rb_alloc_meta_page(struct ring_buffer_per_cpu *cpu_buffer) 6995{ 6996 struct page *page; 6997 6998 if (cpu_buffer->meta_page) 6999 return 0; 7000 7001 page = alloc_page(GFP_USER | __GFP_ZERO); 7002 if (!page) 7003 return -ENOMEM; 7004 7005 cpu_buffer->meta_page = page_to_virt(page); 7006 7007 return 0; 7008} 7009 7010static void rb_free_meta_page(struct ring_buffer_per_cpu *cpu_buffer) 7011{ 7012 unsigned long addr = (unsigned long)cpu_buffer->meta_page; 7013 7014 free_page(addr); 7015 cpu_buffer->meta_page = NULL; 7016} 7017 7018static void rb_setup_ids_meta_page(struct ring_buffer_per_cpu *cpu_buffer, 7019 unsigned long *subbuf_ids) 7020{ 7021 struct trace_buffer_meta *meta = cpu_buffer->meta_page; 7022 unsigned int nr_subbufs = cpu_buffer->nr_pages + 1; 7023 struct buffer_page *first_subbuf, *subbuf; 7024 int cnt = 0; 7025 int id = 0; 7026 7027 id = rb_page_id(cpu_buffer, cpu_buffer->reader_page, id); 7028 subbuf_ids[id++] = (unsigned long)cpu_buffer->reader_page->page; 7029 cnt++; 7030 7031 first_subbuf = subbuf = rb_set_head_page(cpu_buffer); 7032 do { 7033 id = rb_page_id(cpu_buffer, subbuf, id); 7034 7035 if (WARN_ON(id >= nr_subbufs)) 7036 break; 7037 7038 subbuf_ids[id] = (unsigned long)subbuf->page; 7039 7040 rb_inc_page(&subbuf); 7041 id++; 7042 cnt++; 7043 } while (subbuf != first_subbuf); 7044 7045 WARN_ON(cnt != nr_subbufs); 7046 7047 /* install subbuf ID to kern VA translation */ 7048 cpu_buffer->subbuf_ids = subbuf_ids; 7049 7050 meta->meta_struct_len = sizeof(*meta); 7051 meta->nr_subbufs = nr_subbufs; 7052 meta->subbuf_size = cpu_buffer->buffer->subbuf_size + BUF_PAGE_HDR_SIZE; 7053 meta->meta_page_size = meta->subbuf_size; 7054 7055 rb_update_meta_page(cpu_buffer); 7056} 7057 7058static struct ring_buffer_per_cpu * 7059rb_get_mapped_buffer(struct trace_buffer *buffer, int cpu) 7060{ 7061 struct ring_buffer_per_cpu *cpu_buffer; 7062 7063 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 7064 return ERR_PTR(-EINVAL); 7065 7066 cpu_buffer = buffer->buffers[cpu]; 7067 7068 mutex_lock(&cpu_buffer->mapping_lock); 7069 7070 if (!cpu_buffer->user_mapped) { 7071 mutex_unlock(&cpu_buffer->mapping_lock); 7072 return ERR_PTR(-ENODEV); 7073 } 7074 7075 return cpu_buffer; 7076} 7077 7078static void rb_put_mapped_buffer(struct ring_buffer_per_cpu *cpu_buffer) 7079{ 7080 mutex_unlock(&cpu_buffer->mapping_lock); 7081} 7082 7083/* 7084 * Fast-path for rb_buffer_(un)map(). Called whenever the meta-page doesn't need 7085 * to be set-up or torn-down. 7086 */ 7087static int __rb_inc_dec_mapped(struct ring_buffer_per_cpu *cpu_buffer, 7088 bool inc) 7089{ 7090 unsigned long flags; 7091 7092 lockdep_assert_held(&cpu_buffer->mapping_lock); 7093 7094 /* mapped is always greater or equal to user_mapped */ 7095 if (WARN_ON(cpu_buffer->mapped < cpu_buffer->user_mapped)) 7096 return -EINVAL; 7097 7098 if (inc && cpu_buffer->mapped == UINT_MAX) 7099 return -EBUSY; 7100 7101 if (WARN_ON(!inc && cpu_buffer->user_mapped == 0)) 7102 return -EINVAL; 7103 7104 mutex_lock(&cpu_buffer->buffer->mutex); 7105 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 7106 7107 if (inc) { 7108 cpu_buffer->user_mapped++; 7109 cpu_buffer->mapped++; 7110 } else { 7111 cpu_buffer->user_mapped--; 7112 cpu_buffer->mapped--; 7113 } 7114 7115 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 7116 mutex_unlock(&cpu_buffer->buffer->mutex); 7117 7118 return 0; 7119} 7120 7121/* 7122 * +--------------+ pgoff == 0 7123 * | meta page | 7124 * +--------------+ pgoff == 1 7125 * | subbuffer 0 | 7126 * | | 7127 * +--------------+ pgoff == (1 + (1 << subbuf_order)) 7128 * | subbuffer 1 | 7129 * | | 7130 * ... 7131 */ 7132#ifdef CONFIG_MMU 7133static int __rb_map_vma(struct ring_buffer_per_cpu *cpu_buffer, 7134 struct vm_area_struct *vma) 7135{ 7136 unsigned long nr_subbufs, nr_pages, nr_vma_pages, pgoff = vma->vm_pgoff; 7137 unsigned int subbuf_pages, subbuf_order; 7138 struct page **pages __free(kfree) = NULL; 7139 int p = 0, s = 0; 7140 int err; 7141 7142 /* Refuse MP_PRIVATE or writable mappings */ 7143 if (vma->vm_flags & VM_WRITE || vma->vm_flags & VM_EXEC || 7144 !(vma->vm_flags & VM_MAYSHARE)) 7145 return -EPERM; 7146 7147 subbuf_order = cpu_buffer->buffer->subbuf_order; 7148 subbuf_pages = 1 << subbuf_order; 7149 7150 if (subbuf_order && pgoff % subbuf_pages) 7151 return -EINVAL; 7152 7153 /* 7154 * Make sure the mapping cannot become writable later. Also tell the VM 7155 * to not touch these pages (VM_DONTCOPY | VM_DONTEXPAND). 7156 */ 7157 vm_flags_mod(vma, VM_DONTCOPY | VM_DONTEXPAND | VM_DONTDUMP, 7158 VM_MAYWRITE); 7159 7160 lockdep_assert_held(&cpu_buffer->mapping_lock); 7161 7162 nr_subbufs = cpu_buffer->nr_pages + 1; /* + reader-subbuf */ 7163 nr_pages = ((nr_subbufs + 1) << subbuf_order); /* + meta-page */ 7164 if (nr_pages <= pgoff) 7165 return -EINVAL; 7166 7167 nr_pages -= pgoff; 7168 7169 nr_vma_pages = vma_pages(vma); 7170 if (!nr_vma_pages || nr_vma_pages > nr_pages) 7171 return -EINVAL; 7172 7173 nr_pages = nr_vma_pages; 7174 7175 pages = kcalloc(nr_pages, sizeof(*pages), GFP_KERNEL); 7176 if (!pages) 7177 return -ENOMEM; 7178 7179 if (!pgoff) { 7180 unsigned long meta_page_padding; 7181 7182 pages[p++] = virt_to_page(cpu_buffer->meta_page); 7183 7184 /* 7185 * Pad with the zero-page to align the meta-page with the 7186 * sub-buffers. 7187 */ 7188 meta_page_padding = subbuf_pages - 1; 7189 while (meta_page_padding-- && p < nr_pages) { 7190 unsigned long __maybe_unused zero_addr = 7191 vma->vm_start + (PAGE_SIZE * p); 7192 7193 pages[p++] = ZERO_PAGE(zero_addr); 7194 } 7195 } else { 7196 /* Skip the meta-page */ 7197 pgoff -= subbuf_pages; 7198 7199 s += pgoff / subbuf_pages; 7200 } 7201 7202 while (p < nr_pages) { 7203 struct page *page; 7204 int off = 0; 7205 7206 if (WARN_ON_ONCE(s >= nr_subbufs)) 7207 return -EINVAL; 7208 7209 page = virt_to_page((void *)cpu_buffer->subbuf_ids[s]); 7210 7211 for (; off < (1 << (subbuf_order)); off++, page++) { 7212 if (p >= nr_pages) 7213 break; 7214 7215 pages[p++] = page; 7216 } 7217 s++; 7218 } 7219 7220 err = vm_insert_pages(vma, vma->vm_start, pages, &nr_pages); 7221 7222 return err; 7223} 7224#else 7225static int __rb_map_vma(struct ring_buffer_per_cpu *cpu_buffer, 7226 struct vm_area_struct *vma) 7227{ 7228 return -EOPNOTSUPP; 7229} 7230#endif 7231 7232int ring_buffer_map(struct trace_buffer *buffer, int cpu, 7233 struct vm_area_struct *vma) 7234{ 7235 struct ring_buffer_per_cpu *cpu_buffer; 7236 unsigned long flags, *subbuf_ids; 7237 int err; 7238 7239 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 7240 return -EINVAL; 7241 7242 cpu_buffer = buffer->buffers[cpu]; 7243 7244 guard(mutex)(&cpu_buffer->mapping_lock); 7245 7246 if (cpu_buffer->user_mapped) { 7247 err = __rb_map_vma(cpu_buffer, vma); 7248 if (!err) 7249 err = __rb_inc_dec_mapped(cpu_buffer, true); 7250 return err; 7251 } 7252 7253 /* prevent another thread from changing buffer/sub-buffer sizes */ 7254 guard(mutex)(&buffer->mutex); 7255 7256 err = rb_alloc_meta_page(cpu_buffer); 7257 if (err) 7258 return err; 7259 7260 /* subbuf_ids include the reader while nr_pages does not */ 7261 subbuf_ids = kcalloc(cpu_buffer->nr_pages + 1, sizeof(*subbuf_ids), GFP_KERNEL); 7262 if (!subbuf_ids) { 7263 rb_free_meta_page(cpu_buffer); 7264 return -ENOMEM; 7265 } 7266 7267 atomic_inc(&cpu_buffer->resize_disabled); 7268 7269 /* 7270 * Lock all readers to block any subbuf swap until the subbuf IDs are 7271 * assigned. 7272 */ 7273 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 7274 rb_setup_ids_meta_page(cpu_buffer, subbuf_ids); 7275 7276 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 7277 7278 err = __rb_map_vma(cpu_buffer, vma); 7279 if (!err) { 7280 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 7281 /* This is the first time it is mapped by user */ 7282 cpu_buffer->mapped++; 7283 cpu_buffer->user_mapped = 1; 7284 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 7285 } else { 7286 kfree(cpu_buffer->subbuf_ids); 7287 cpu_buffer->subbuf_ids = NULL; 7288 rb_free_meta_page(cpu_buffer); 7289 atomic_dec(&cpu_buffer->resize_disabled); 7290 } 7291 7292 return err; 7293} 7294 7295/* 7296 * This is called when a VMA is duplicated (e.g., on fork()) to increment 7297 * the user_mapped counter without remapping pages. 7298 */ 7299void ring_buffer_map_dup(struct trace_buffer *buffer, int cpu) 7300{ 7301 struct ring_buffer_per_cpu *cpu_buffer; 7302 7303 if (WARN_ON(!cpumask_test_cpu(cpu, buffer->cpumask))) 7304 return; 7305 7306 cpu_buffer = buffer->buffers[cpu]; 7307 7308 guard(mutex)(&cpu_buffer->mapping_lock); 7309 7310 if (cpu_buffer->user_mapped) 7311 __rb_inc_dec_mapped(cpu_buffer, true); 7312 else 7313 WARN(1, "Unexpected buffer stat, it should be mapped"); 7314} 7315 7316int ring_buffer_unmap(struct trace_buffer *buffer, int cpu) 7317{ 7318 struct ring_buffer_per_cpu *cpu_buffer; 7319 unsigned long flags; 7320 7321 if (!cpumask_test_cpu(cpu, buffer->cpumask)) 7322 return -EINVAL; 7323 7324 cpu_buffer = buffer->buffers[cpu]; 7325 7326 guard(mutex)(&cpu_buffer->mapping_lock); 7327 7328 if (!cpu_buffer->user_mapped) { 7329 return -ENODEV; 7330 } else if (cpu_buffer->user_mapped > 1) { 7331 __rb_inc_dec_mapped(cpu_buffer, false); 7332 return 0; 7333 } 7334 7335 guard(mutex)(&buffer->mutex); 7336 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 7337 7338 /* This is the last user space mapping */ 7339 if (!WARN_ON_ONCE(cpu_buffer->mapped < cpu_buffer->user_mapped)) 7340 cpu_buffer->mapped--; 7341 cpu_buffer->user_mapped = 0; 7342 7343 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 7344 7345 kfree(cpu_buffer->subbuf_ids); 7346 cpu_buffer->subbuf_ids = NULL; 7347 rb_free_meta_page(cpu_buffer); 7348 atomic_dec(&cpu_buffer->resize_disabled); 7349 7350 return 0; 7351} 7352 7353int ring_buffer_map_get_reader(struct trace_buffer *buffer, int cpu) 7354{ 7355 struct ring_buffer_per_cpu *cpu_buffer; 7356 struct buffer_page *reader; 7357 unsigned long missed_events; 7358 unsigned long reader_size; 7359 unsigned long flags; 7360 7361 cpu_buffer = rb_get_mapped_buffer(buffer, cpu); 7362 if (IS_ERR(cpu_buffer)) 7363 return (int)PTR_ERR(cpu_buffer); 7364 7365 raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); 7366 7367consume: 7368 if (rb_per_cpu_empty(cpu_buffer)) 7369 goto out; 7370 7371 reader_size = rb_page_size(cpu_buffer->reader_page); 7372 7373 /* 7374 * There are data to be read on the current reader page, we can 7375 * return to the caller. But before that, we assume the latter will read 7376 * everything. Let's update the kernel reader accordingly. 7377 */ 7378 if (cpu_buffer->reader_page->read < reader_size) { 7379 while (cpu_buffer->reader_page->read < reader_size) 7380 rb_advance_reader(cpu_buffer); 7381 goto out; 7382 } 7383 7384 /* Did the reader catch up with the writer? */ 7385 if (cpu_buffer->reader_page == cpu_buffer->commit_page) 7386 goto out; 7387 7388 reader = rb_get_reader_page(cpu_buffer); 7389 if (WARN_ON(!reader)) 7390 goto out; 7391 7392 /* Check if any events were dropped */ 7393 missed_events = cpu_buffer->lost_events; 7394 7395 if (missed_events) { 7396 if (cpu_buffer->reader_page != cpu_buffer->commit_page) { 7397 struct buffer_data_page *bpage = reader->page; 7398 unsigned int commit; 7399 /* 7400 * Use the real_end for the data size, 7401 * This gives us a chance to store the lost events 7402 * on the page. 7403 */ 7404 if (reader->real_end) 7405 local_set(&bpage->commit, reader->real_end); 7406 /* 7407 * If there is room at the end of the page to save the 7408 * missed events, then record it there. 7409 */ 7410 commit = rb_page_size(reader); 7411 if (buffer->subbuf_size - commit >= sizeof(missed_events)) { 7412 memcpy(&bpage->data[commit], &missed_events, 7413 sizeof(missed_events)); 7414 local_add(RB_MISSED_STORED, &bpage->commit); 7415 } 7416 local_add(RB_MISSED_EVENTS, &bpage->commit); 7417 } else if (!WARN_ONCE(cpu_buffer->reader_page == cpu_buffer->tail_page, 7418 "Reader on commit with %ld missed events", 7419 missed_events)) { 7420 /* 7421 * There shouldn't be any missed events if the tail_page 7422 * is on the reader page. But if the tail page is not on the 7423 * reader page and the commit_page is, that would mean that 7424 * there's a commit_overrun (an interrupt preempted an 7425 * addition of an event and then filled the buffer 7426 * with new events). In this case it's not an 7427 * error, but it should still be reported. 7428 * 7429 * TODO: Add missed events to the page for user space to know. 7430 */ 7431 pr_info("Ring buffer [%d] commit overrun lost %ld events at timestamp:%lld\n", 7432 cpu, missed_events, cpu_buffer->reader_page->page->time_stamp); 7433 } 7434 } 7435 7436 cpu_buffer->lost_events = 0; 7437 7438 goto consume; 7439 7440out: 7441 /* Some archs do not have data cache coherency between kernel and user-space */ 7442 flush_kernel_vmap_range(cpu_buffer->reader_page->page, 7443 buffer->subbuf_size + BUF_PAGE_HDR_SIZE); 7444 7445 rb_update_meta_page(cpu_buffer); 7446 7447 raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); 7448 rb_put_mapped_buffer(cpu_buffer); 7449 7450 return 0; 7451} 7452 7453/* 7454 * We only allocate new buffers, never free them if the CPU goes down. 7455 * If we were to free the buffer, then the user would lose any trace that was in 7456 * the buffer. 7457 */ 7458int trace_rb_cpu_prepare(unsigned int cpu, struct hlist_node *node) 7459{ 7460 struct trace_buffer *buffer; 7461 long nr_pages_same; 7462 int cpu_i; 7463 unsigned long nr_pages; 7464 7465 buffer = container_of(node, struct trace_buffer, node); 7466 if (cpumask_test_cpu(cpu, buffer->cpumask)) 7467 return 0; 7468 7469 nr_pages = 0; 7470 nr_pages_same = 1; 7471 /* check if all cpu sizes are same */ 7472 for_each_buffer_cpu(buffer, cpu_i) { 7473 /* fill in the size from first enabled cpu */ 7474 if (nr_pages == 0) 7475 nr_pages = buffer->buffers[cpu_i]->nr_pages; 7476 if (nr_pages != buffer->buffers[cpu_i]->nr_pages) { 7477 nr_pages_same = 0; 7478 break; 7479 } 7480 } 7481 /* allocate minimum pages, user can later expand it */ 7482 if (!nr_pages_same) 7483 nr_pages = 2; 7484 buffer->buffers[cpu] = 7485 rb_allocate_cpu_buffer(buffer, nr_pages, cpu); 7486 if (!buffer->buffers[cpu]) { 7487 WARN(1, "failed to allocate ring buffer on CPU %u\n", 7488 cpu); 7489 return -ENOMEM; 7490 } 7491 smp_wmb(); 7492 cpumask_set_cpu(cpu, buffer->cpumask); 7493 return 0; 7494} 7495 7496#ifdef CONFIG_RING_BUFFER_STARTUP_TEST 7497/* 7498 * This is a basic integrity check of the ring buffer. 7499 * Late in the boot cycle this test will run when configured in. 7500 * It will kick off a thread per CPU that will go into a loop 7501 * writing to the per cpu ring buffer various sizes of data. 7502 * Some of the data will be large items, some small. 7503 * 7504 * Another thread is created that goes into a spin, sending out 7505 * IPIs to the other CPUs to also write into the ring buffer. 7506 * this is to test the nesting ability of the buffer. 7507 * 7508 * Basic stats are recorded and reported. If something in the 7509 * ring buffer should happen that's not expected, a big warning 7510 * is displayed and all ring buffers are disabled. 7511 */ 7512static struct task_struct *rb_threads[NR_CPUS] __initdata; 7513 7514struct rb_test_data { 7515 struct trace_buffer *buffer; 7516 unsigned long events; 7517 unsigned long bytes_written; 7518 unsigned long bytes_alloc; 7519 unsigned long bytes_dropped; 7520 unsigned long events_nested; 7521 unsigned long bytes_written_nested; 7522 unsigned long bytes_alloc_nested; 7523 unsigned long bytes_dropped_nested; 7524 int min_size_nested; 7525 int max_size_nested; 7526 int max_size; 7527 int min_size; 7528 int cpu; 7529 int cnt; 7530}; 7531 7532static struct rb_test_data rb_data[NR_CPUS] __initdata; 7533 7534/* 1 meg per cpu */ 7535#define RB_TEST_BUFFER_SIZE 1048576 7536 7537static char rb_string[] __initdata = 7538 "abcdefghijklmnopqrstuvwxyz1234567890!@#$%^&*()?+\\" 7539 "?+|:';\",.<>/?abcdefghijklmnopqrstuvwxyz1234567890" 7540 "!@#$%^&*()?+\\?+|:';\",.<>/?abcdefghijklmnopqrstuv"; 7541 7542static bool rb_test_started __initdata; 7543 7544struct rb_item { 7545 int size; 7546 char str[]; 7547}; 7548 7549static __init int rb_write_something(struct rb_test_data *data, bool nested) 7550{ 7551 struct ring_buffer_event *event; 7552 struct rb_item *item; 7553 bool started; 7554 int event_len; 7555 int size; 7556 int len; 7557 int cnt; 7558 7559 /* Have nested writes different that what is written */ 7560 cnt = data->cnt + (nested ? 27 : 0); 7561 7562 /* Multiply cnt by ~e, to make some unique increment */ 7563 size = (cnt * 68 / 25) % (sizeof(rb_string) - 1); 7564 7565 len = size + sizeof(struct rb_item); 7566 7567 started = rb_test_started; 7568 /* read rb_test_started before checking buffer enabled */ 7569 smp_rmb(); 7570 7571 event = ring_buffer_lock_reserve(data->buffer, len); 7572 if (!event) { 7573 /* Ignore dropped events before test starts. */ 7574 if (started) { 7575 if (nested) 7576 data->bytes_dropped_nested += len; 7577 else 7578 data->bytes_dropped += len; 7579 } 7580 return len; 7581 } 7582 7583 event_len = ring_buffer_event_length(event); 7584 7585 if (RB_WARN_ON(data->buffer, event_len < len)) 7586 goto out; 7587 7588 item = ring_buffer_event_data(event); 7589 item->size = size; 7590 memcpy(item->str, rb_string, size); 7591 7592 if (nested) { 7593 data->bytes_alloc_nested += event_len; 7594 data->bytes_written_nested += len; 7595 data->events_nested++; 7596 if (!data->min_size_nested || len < data->min_size_nested) 7597 data->min_size_nested = len; 7598 if (len > data->max_size_nested) 7599 data->max_size_nested = len; 7600 } else { 7601 data->bytes_alloc += event_len; 7602 data->bytes_written += len; 7603 data->events++; 7604 if (!data->min_size || len < data->min_size) 7605 data->max_size = len; 7606 if (len > data->max_size) 7607 data->max_size = len; 7608 } 7609 7610 out: 7611 ring_buffer_unlock_commit(data->buffer); 7612 7613 return 0; 7614} 7615 7616static __init int rb_test(void *arg) 7617{ 7618 struct rb_test_data *data = arg; 7619 7620 while (!kthread_should_stop()) { 7621 rb_write_something(data, false); 7622 data->cnt++; 7623 7624 set_current_state(TASK_INTERRUPTIBLE); 7625 /* Now sleep between a min of 100-300us and a max of 1ms */ 7626 usleep_range(((data->cnt % 3) + 1) * 100, 1000); 7627 } 7628 7629 return 0; 7630} 7631 7632static __init void rb_ipi(void *ignore) 7633{ 7634 struct rb_test_data *data; 7635 int cpu = smp_processor_id(); 7636 7637 data = &rb_data[cpu]; 7638 rb_write_something(data, true); 7639} 7640 7641static __init int rb_hammer_test(void *arg) 7642{ 7643 while (!kthread_should_stop()) { 7644 7645 /* Send an IPI to all cpus to write data! */ 7646 smp_call_function(rb_ipi, NULL, 1); 7647 /* No sleep, but for non preempt, let others run */ 7648 schedule(); 7649 } 7650 7651 return 0; 7652} 7653 7654static __init int test_ringbuffer(void) 7655{ 7656 struct task_struct *rb_hammer; 7657 struct trace_buffer *buffer; 7658 int cpu; 7659 int ret = 0; 7660 7661 if (security_locked_down(LOCKDOWN_TRACEFS)) { 7662 pr_warn("Lockdown is enabled, skipping ring buffer tests\n"); 7663 return 0; 7664 } 7665 7666 pr_info("Running ring buffer tests...\n"); 7667 7668 buffer = ring_buffer_alloc(RB_TEST_BUFFER_SIZE, RB_FL_OVERWRITE); 7669 if (WARN_ON(!buffer)) 7670 return 0; 7671 7672 /* Disable buffer so that threads can't write to it yet */ 7673 ring_buffer_record_off(buffer); 7674 7675 for_each_online_cpu(cpu) { 7676 rb_data[cpu].buffer = buffer; 7677 rb_data[cpu].cpu = cpu; 7678 rb_data[cpu].cnt = cpu; 7679 rb_threads[cpu] = kthread_run_on_cpu(rb_test, &rb_data[cpu], 7680 cpu, "rbtester/%u"); 7681 if (WARN_ON(IS_ERR(rb_threads[cpu]))) { 7682 pr_cont("FAILED\n"); 7683 ret = PTR_ERR(rb_threads[cpu]); 7684 goto out_free; 7685 } 7686 } 7687 7688 /* Now create the rb hammer! */ 7689 rb_hammer = kthread_run(rb_hammer_test, NULL, "rbhammer"); 7690 if (WARN_ON(IS_ERR(rb_hammer))) { 7691 pr_cont("FAILED\n"); 7692 ret = PTR_ERR(rb_hammer); 7693 goto out_free; 7694 } 7695 7696 ring_buffer_record_on(buffer); 7697 /* 7698 * Show buffer is enabled before setting rb_test_started. 7699 * Yes there's a small race window where events could be 7700 * dropped and the thread won't catch it. But when a ring 7701 * buffer gets enabled, there will always be some kind of 7702 * delay before other CPUs see it. Thus, we don't care about 7703 * those dropped events. We care about events dropped after 7704 * the threads see that the buffer is active. 7705 */ 7706 smp_wmb(); 7707 rb_test_started = true; 7708 7709 set_current_state(TASK_INTERRUPTIBLE); 7710 /* Just run for 10 seconds */ 7711 schedule_timeout(10 * HZ); 7712 7713 kthread_stop(rb_hammer); 7714 7715 out_free: 7716 for_each_online_cpu(cpu) { 7717 if (!rb_threads[cpu]) 7718 break; 7719 kthread_stop(rb_threads[cpu]); 7720 } 7721 if (ret) { 7722 ring_buffer_free(buffer); 7723 return ret; 7724 } 7725 7726 /* Report! */ 7727 pr_info("finished\n"); 7728 for_each_online_cpu(cpu) { 7729 struct ring_buffer_event *event; 7730 struct rb_test_data *data = &rb_data[cpu]; 7731 struct rb_item *item; 7732 unsigned long total_events; 7733 unsigned long total_dropped; 7734 unsigned long total_written; 7735 unsigned long total_alloc; 7736 unsigned long total_read = 0; 7737 unsigned long total_size = 0; 7738 unsigned long total_len = 0; 7739 unsigned long total_lost = 0; 7740 unsigned long lost; 7741 int big_event_size; 7742 int small_event_size; 7743 7744 ret = -1; 7745 7746 total_events = data->events + data->events_nested; 7747 total_written = data->bytes_written + data->bytes_written_nested; 7748 total_alloc = data->bytes_alloc + data->bytes_alloc_nested; 7749 total_dropped = data->bytes_dropped + data->bytes_dropped_nested; 7750 7751 big_event_size = data->max_size + data->max_size_nested; 7752 small_event_size = data->min_size + data->min_size_nested; 7753 7754 pr_info("CPU %d:\n", cpu); 7755 pr_info(" events: %ld\n", total_events); 7756 pr_info(" dropped bytes: %ld\n", total_dropped); 7757 pr_info(" alloced bytes: %ld\n", total_alloc); 7758 pr_info(" written bytes: %ld\n", total_written); 7759 pr_info(" biggest event: %d\n", big_event_size); 7760 pr_info(" smallest event: %d\n", small_event_size); 7761 7762 if (RB_WARN_ON(buffer, total_dropped)) 7763 break; 7764 7765 ret = 0; 7766 7767 while ((event = ring_buffer_consume(buffer, cpu, NULL, &lost))) { 7768 total_lost += lost; 7769 item = ring_buffer_event_data(event); 7770 total_len += ring_buffer_event_length(event); 7771 total_size += item->size + sizeof(struct rb_item); 7772 if (memcmp(&item->str[0], rb_string, item->size) != 0) { 7773 pr_info("FAILED!\n"); 7774 pr_info("buffer had: %.*s\n", item->size, item->str); 7775 pr_info("expected: %.*s\n", item->size, rb_string); 7776 RB_WARN_ON(buffer, 1); 7777 ret = -1; 7778 break; 7779 } 7780 total_read++; 7781 } 7782 if (ret) 7783 break; 7784 7785 ret = -1; 7786 7787 pr_info(" read events: %ld\n", total_read); 7788 pr_info(" lost events: %ld\n", total_lost); 7789 pr_info(" total events: %ld\n", total_lost + total_read); 7790 pr_info(" recorded len bytes: %ld\n", total_len); 7791 pr_info(" recorded size bytes: %ld\n", total_size); 7792 if (total_lost) { 7793 pr_info(" With dropped events, record len and size may not match\n" 7794 " alloced and written from above\n"); 7795 } else { 7796 if (RB_WARN_ON(buffer, total_len != total_alloc || 7797 total_size != total_written)) 7798 break; 7799 } 7800 if (RB_WARN_ON(buffer, total_lost + total_read != total_events)) 7801 break; 7802 7803 ret = 0; 7804 } 7805 if (!ret) 7806 pr_info("Ring buffer PASSED!\n"); 7807 7808 ring_buffer_free(buffer); 7809 return 0; 7810} 7811 7812late_initcall(test_ringbuffer); 7813#endif /* CONFIG_RING_BUFFER_STARTUP_TEST */
~/
Read-only file preview.