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

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

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

~ [ 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