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

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

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

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