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