~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

TOMOYO Linux Cross Reference
Linux/kernel/trace/ring_buffer.c

Version: ~ [ linux-5.14-rc3 ] ~ [ linux-5.13.5 ] ~ [ linux-5.12.19 ] ~ [ linux-5.11.22 ] ~ [ linux-5.10.53 ] ~ [ linux-5.9.16 ] ~ [ linux-5.8.18 ] ~ [ linux-5.7.19 ] ~ [ linux-5.6.19 ] ~ [ linux-5.5.19 ] ~ [ linux-5.4.135 ] ~ [ linux-5.3.18 ] ~ [ linux-5.2.21 ] ~ [ linux-5.1.21 ] ~ [ linux-5.0.21 ] ~ [ linux-4.20.17 ] ~ [ linux-4.19.198 ] ~ [ linux-4.18.20 ] ~ [ linux-4.17.19 ] ~ [ linux-4.16.18 ] ~ [ linux-4.15.18 ] ~ [ linux-4.14.240 ] ~ [ linux-4.13.16 ] ~ [ linux-4.12.14 ] ~ [ linux-4.11.12 ] ~ [ linux-4.10.17 ] ~ [ linux-4.9.276 ] ~ [ linux-4.8.17 ] ~ [ linux-4.7.10 ] ~ [ linux-4.6.7 ] ~ [ linux-4.5.7 ] ~ [ linux-4.4.276 ] ~ [ linux-4.3.6 ] ~ [ linux-4.2.8 ] ~ [ linux-4.1.52 ] ~ [ linux-4.0.9 ] ~ [ linux-3.18.140 ] ~ [ linux-3.16.85 ] ~ [ linux-3.14.79 ] ~ [ linux-3.12.74 ] ~ [ linux-3.10.108 ] ~ [ linux-2.6.32.71 ] ~ [ linux-2.6.0 ] ~ [ linux-2.4.37.11 ] ~ [ unix-v6-master ] ~ [ ccs-tools-1.8.5 ] ~ [ policy-sample ] ~
Architecture: ~ [ i386 ] ~ [ alpha ] ~ [ m68k ] ~ [ mips ] ~ [ ppc ] ~ [ sparc ] ~ [ sparc64 ] ~

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

~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

kernel.org | git.kernel.org | LWN.net | Project Home | Wiki (Japanese) | Wiki (English) | SVN repository | Mail admin

Linux® is a registered trademark of Linus Torvalds in the United States and other countries.
TOMOYO® is a registered trademark of NTT DATA CORPORATION.

osdn.jp