kernel/trace/ring_buffer.c

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