xref: /linux/kernel/trace/ring_buffer.c (revision ef9226cd56b718c79184a3466d32984a51cb449c)
1 // SPDX-License-Identifier: GPL-2.0
2 /*
3  * Generic ring buffer
4  *
5  * Copyright (C) 2008 Steven Rostedt <srostedt@redhat.com>
6  */
7 #include <linux/trace_recursion.h>
8 #include <linux/trace_events.h>
9 #include <linux/ring_buffer.h>
10 #include <linux/trace_clock.h>
11 #include <linux/sched/clock.h>
12 #include <linux/trace_seq.h>
13 #include <linux/spinlock.h>
14 #include <linux/irq_work.h>
15 #include <linux/security.h>
16 #include <linux/uaccess.h>
17 #include <linux/hardirq.h>
18 #include <linux/kthread.h>	/* for self test */
19 #include <linux/module.h>
20 #include <linux/percpu.h>
21 #include <linux/mutex.h>
22 #include <linux/delay.h>
23 #include <linux/slab.h>
24 #include <linux/init.h>
25 #include <linux/hash.h>
26 #include <linux/list.h>
27 #include <linux/cpu.h>
28 #include <linux/oom.h>
29 
30 #include <asm/local64.h>
31 #include <asm/local.h>
32 
33 /*
34  * The "absolute" timestamp in the buffer is only 59 bits.
35  * If a clock has the 5 MSBs set, it needs to be saved and
36  * reinserted.
37  */
38 #define TS_MSB		(0xf8ULL << 56)
39 #define ABS_TS_MASK	(~TS_MSB)
40 
41 static void update_pages_handler(struct work_struct *work);
42 
43 /*
44  * The ring buffer header is special. We must manually up keep it.
45  */
46 int ring_buffer_print_entry_header(struct trace_seq *s)
47 {
48 	trace_seq_puts(s, "# compressed entry header\n");
49 	trace_seq_puts(s, "\ttype_len    :    5 bits\n");
50 	trace_seq_puts(s, "\ttime_delta  :   27 bits\n");
51 	trace_seq_puts(s, "\tarray       :   32 bits\n");
52 	trace_seq_putc(s, '\n');
53 	trace_seq_printf(s, "\tpadding     : type == %d\n",
54 			 RINGBUF_TYPE_PADDING);
55 	trace_seq_printf(s, "\ttime_extend : type == %d\n",
56 			 RINGBUF_TYPE_TIME_EXTEND);
57 	trace_seq_printf(s, "\ttime_stamp : type == %d\n",
58 			 RINGBUF_TYPE_TIME_STAMP);
59 	trace_seq_printf(s, "\tdata max type_len  == %d\n",
60 			 RINGBUF_TYPE_DATA_TYPE_LEN_MAX);
61 
62 	return !trace_seq_has_overflowed(s);
63 }
64 
65 /*
66  * The ring buffer is made up of a list of pages. A separate list of pages is
67  * allocated for each CPU. A writer may only write to a buffer that is
68  * associated with the CPU it is currently executing on.  A reader may read
69  * from any per cpu buffer.
70  *
71  * The reader is special. For each per cpu buffer, the reader has its own
72  * reader page. When a reader has read the entire reader page, this reader
73  * page is swapped with another page in the ring buffer.
74  *
75  * Now, as long as the writer is off the reader page, the reader can do what
76  * ever it wants with that page. The writer will never write to that page
77  * again (as long as it is out of the ring buffer).
78  *
79  * Here's some silly ASCII art.
80  *
81  *   +------+
82  *   |reader|          RING BUFFER
83  *   |page  |
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  *   |reader|          RING BUFFER
105  *   |page  |------------------v
106  *   +------+        +---+   +---+   +---+
107  *      ^            |   |-->|   |-->|   |
108  *      |            +---+   +---+   +---+
109  *      |                              |
110  *      |                              |
111  *      +------------------------------+
112  *
113  *
114  *   +------+
115  *   |buffer|          RING BUFFER
116  *   |page  |------------------v
117  *   +------+        +---+   +---+   +---+
118  *      ^            |   |   |   |-->|   |
119  *      |   New      +---+   +---+   +---+
120  *      |  Reader------^               |
121  *      |   page                       |
122  *      +------------------------------+
123  *
124  *
125  * After we make this swap, the reader can hand this page off to the splice
126  * code and be done with it. It can even allocate a new page if it needs to
127  * and swap that into the ring buffer.
128  *
129  * We will be using cmpxchg soon to make all this lockless.
130  *
131  */
132 
133 /* Used for individual buffers (after the counter) */
134 #define RB_BUFFER_OFF		(1 << 20)
135 
136 #define BUF_PAGE_HDR_SIZE offsetof(struct buffer_data_page, data)
137 
138 #define RB_EVNT_HDR_SIZE (offsetof(struct ring_buffer_event, array))
139 #define RB_ALIGNMENT		4U
140 #define RB_MAX_SMALL_DATA	(RB_ALIGNMENT * RINGBUF_TYPE_DATA_TYPE_LEN_MAX)
141 #define RB_EVNT_MIN_SIZE	8U	/* two 32bit words */
142 
143 #ifndef CONFIG_HAVE_64BIT_ALIGNED_ACCESS
144 # define RB_FORCE_8BYTE_ALIGNMENT	0
145 # define RB_ARCH_ALIGNMENT		RB_ALIGNMENT
146 #else
147 # define RB_FORCE_8BYTE_ALIGNMENT	1
148 # define RB_ARCH_ALIGNMENT		8U
149 #endif
150 
151 #define RB_ALIGN_DATA		__aligned(RB_ARCH_ALIGNMENT)
152 
153 /* define RINGBUF_TYPE_DATA for 'case RINGBUF_TYPE_DATA:' */
154 #define RINGBUF_TYPE_DATA 0 ... RINGBUF_TYPE_DATA_TYPE_LEN_MAX
155 
156 enum {
157 	RB_LEN_TIME_EXTEND = 8,
158 	RB_LEN_TIME_STAMP =  8,
159 };
160 
161 #define skip_time_extend(event) \
162 	((struct ring_buffer_event *)((char *)event + RB_LEN_TIME_EXTEND))
163 
164 #define extended_time(event) \
165 	(event->type_len >= RINGBUF_TYPE_TIME_EXTEND)
166 
167 static inline bool rb_null_event(struct ring_buffer_event *event)
168 {
169 	return event->type_len == RINGBUF_TYPE_PADDING && !event->time_delta;
170 }
171 
172 static void rb_event_set_padding(struct ring_buffer_event *event)
173 {
174 	/* padding has a NULL time_delta */
175 	event->type_len = RINGBUF_TYPE_PADDING;
176 	event->time_delta = 0;
177 }
178 
179 static unsigned
180 rb_event_data_length(struct ring_buffer_event *event)
181 {
182 	unsigned length;
183 
184 	if (event->type_len)
185 		length = event->type_len * RB_ALIGNMENT;
186 	else
187 		length = event->array[0];
188 	return length + RB_EVNT_HDR_SIZE;
189 }
190 
191 /*
192  * Return the length of the given event. Will return
193  * the length of the time extend if the event is a
194  * time extend.
195  */
196 static inline unsigned
197 rb_event_length(struct ring_buffer_event *event)
198 {
199 	switch (event->type_len) {
200 	case RINGBUF_TYPE_PADDING:
201 		if (rb_null_event(event))
202 			/* undefined */
203 			return -1;
204 		return  event->array[0] + RB_EVNT_HDR_SIZE;
205 
206 	case RINGBUF_TYPE_TIME_EXTEND:
207 		return RB_LEN_TIME_EXTEND;
208 
209 	case RINGBUF_TYPE_TIME_STAMP:
210 		return RB_LEN_TIME_STAMP;
211 
212 	case RINGBUF_TYPE_DATA:
213 		return rb_event_data_length(event);
214 	default:
215 		WARN_ON_ONCE(1);
216 	}
217 	/* not hit */
218 	return 0;
219 }
220 
221 /*
222  * Return total length of time extend and data,
223  *   or just the event length for all other events.
224  */
225 static inline unsigned
226 rb_event_ts_length(struct ring_buffer_event *event)
227 {
228 	unsigned len = 0;
229 
230 	if (extended_time(event)) {
231 		/* time extends include the data event after it */
232 		len = RB_LEN_TIME_EXTEND;
233 		event = skip_time_extend(event);
234 	}
235 	return len + rb_event_length(event);
236 }
237 
238 /**
239  * ring_buffer_event_length - return the length of the event
240  * @event: the event to get the length of
241  *
242  * Returns the size of the data load of a data event.
243  * If the event is something other than a data event, it
244  * returns the size of the event itself. With the exception
245  * of a TIME EXTEND, where it still returns the size of the
246  * data load of the data event after it.
247  */
248 unsigned ring_buffer_event_length(struct ring_buffer_event *event)
249 {
250 	unsigned length;
251 
252 	if (extended_time(event))
253 		event = skip_time_extend(event);
254 
255 	length = rb_event_length(event);
256 	if (event->type_len > RINGBUF_TYPE_DATA_TYPE_LEN_MAX)
257 		return length;
258 	length -= RB_EVNT_HDR_SIZE;
259 	if (length > RB_MAX_SMALL_DATA + sizeof(event->array[0]))
260                 length -= sizeof(event->array[0]);
261 	return length;
262 }
263 EXPORT_SYMBOL_GPL(ring_buffer_event_length);
264 
265 /* inline for ring buffer fast paths */
266 static __always_inline void *
267 rb_event_data(struct ring_buffer_event *event)
268 {
269 	if (extended_time(event))
270 		event = skip_time_extend(event);
271 	WARN_ON_ONCE(event->type_len > RINGBUF_TYPE_DATA_TYPE_LEN_MAX);
272 	/* If length is in len field, then array[0] has the data */
273 	if (event->type_len)
274 		return (void *)&event->array[0];
275 	/* Otherwise length is in array[0] and array[1] has the data */
276 	return (void *)&event->array[1];
277 }
278 
279 /**
280  * ring_buffer_event_data - return the data of the event
281  * @event: the event to get the data from
282  */
283 void *ring_buffer_event_data(struct ring_buffer_event *event)
284 {
285 	return rb_event_data(event);
286 }
287 EXPORT_SYMBOL_GPL(ring_buffer_event_data);
288 
289 #define for_each_buffer_cpu(buffer, cpu)		\
290 	for_each_cpu(cpu, buffer->cpumask)
291 
292 #define for_each_online_buffer_cpu(buffer, cpu)		\
293 	for_each_cpu_and(cpu, buffer->cpumask, cpu_online_mask)
294 
295 #define TS_SHIFT	27
296 #define TS_MASK		((1ULL << TS_SHIFT) - 1)
297 #define TS_DELTA_TEST	(~TS_MASK)
298 
299 static u64 rb_event_time_stamp(struct ring_buffer_event *event)
300 {
301 	u64 ts;
302 
303 	ts = event->array[0];
304 	ts <<= TS_SHIFT;
305 	ts += event->time_delta;
306 
307 	return ts;
308 }
309 
310 /* Flag when events were overwritten */
311 #define RB_MISSED_EVENTS	(1 << 31)
312 /* Missed count stored at end */
313 #define RB_MISSED_STORED	(1 << 30)
314 
315 struct buffer_data_page {
316 	u64		 time_stamp;	/* page time stamp */
317 	local_t		 commit;	/* write committed index */
318 	unsigned char	 data[] RB_ALIGN_DATA;	/* data of buffer page */
319 };
320 
321 struct buffer_data_read_page {
322 	unsigned		order;	/* order of the page */
323 	struct buffer_data_page	*data;	/* actual data, stored in this page */
324 };
325 
326 /*
327  * Note, the buffer_page list must be first. The buffer pages
328  * are allocated in cache lines, which means that each buffer
329  * page will be at the beginning of a cache line, and thus
330  * the least significant bits will be zero. We use this to
331  * add flags in the list struct pointers, to make the ring buffer
332  * lockless.
333  */
334 struct buffer_page {
335 	struct list_head list;		/* list of buffer pages */
336 	local_t		 write;		/* index for next write */
337 	unsigned	 read;		/* index for next read */
338 	local_t		 entries;	/* entries on this page */
339 	unsigned long	 real_end;	/* real end of data */
340 	unsigned	 order;		/* order of the page */
341 	struct buffer_data_page *page;	/* Actual data page */
342 };
343 
344 /*
345  * The buffer page counters, write and entries, must be reset
346  * atomically when crossing page boundaries. To synchronize this
347  * update, two counters are inserted into the number. One is
348  * the actual counter for the write position or count on the page.
349  *
350  * The other is a counter of updaters. Before an update happens
351  * the update partition of the counter is incremented. This will
352  * allow the updater to update the counter atomically.
353  *
354  * The counter is 20 bits, and the state data is 12.
355  */
356 #define RB_WRITE_MASK		0xfffff
357 #define RB_WRITE_INTCNT		(1 << 20)
358 
359 static void rb_init_page(struct buffer_data_page *bpage)
360 {
361 	local_set(&bpage->commit, 0);
362 }
363 
364 static __always_inline unsigned int rb_page_commit(struct buffer_page *bpage)
365 {
366 	return local_read(&bpage->page->commit);
367 }
368 
369 static void free_buffer_page(struct buffer_page *bpage)
370 {
371 	free_pages((unsigned long)bpage->page, bpage->order);
372 	kfree(bpage);
373 }
374 
375 /*
376  * We need to fit the time_stamp delta into 27 bits.
377  */
378 static inline bool test_time_stamp(u64 delta)
379 {
380 	return !!(delta & TS_DELTA_TEST);
381 }
382 
383 struct rb_irq_work {
384 	struct irq_work			work;
385 	wait_queue_head_t		waiters;
386 	wait_queue_head_t		full_waiters;
387 	atomic_t			seq;
388 	bool				waiters_pending;
389 	bool				full_waiters_pending;
390 	bool				wakeup_full;
391 };
392 
393 /*
394  * Structure to hold event state and handle nested events.
395  */
396 struct rb_event_info {
397 	u64			ts;
398 	u64			delta;
399 	u64			before;
400 	u64			after;
401 	unsigned long		length;
402 	struct buffer_page	*tail_page;
403 	int			add_timestamp;
404 };
405 
406 /*
407  * Used for the add_timestamp
408  *  NONE
409  *  EXTEND - wants a time extend
410  *  ABSOLUTE - the buffer requests all events to have absolute time stamps
411  *  FORCE - force a full time stamp.
412  */
413 enum {
414 	RB_ADD_STAMP_NONE		= 0,
415 	RB_ADD_STAMP_EXTEND		= BIT(1),
416 	RB_ADD_STAMP_ABSOLUTE		= BIT(2),
417 	RB_ADD_STAMP_FORCE		= BIT(3)
418 };
419 /*
420  * Used for which event context the event is in.
421  *  TRANSITION = 0
422  *  NMI     = 1
423  *  IRQ     = 2
424  *  SOFTIRQ = 3
425  *  NORMAL  = 4
426  *
427  * See trace_recursive_lock() comment below for more details.
428  */
429 enum {
430 	RB_CTX_TRANSITION,
431 	RB_CTX_NMI,
432 	RB_CTX_IRQ,
433 	RB_CTX_SOFTIRQ,
434 	RB_CTX_NORMAL,
435 	RB_CTX_MAX
436 };
437 
438 struct rb_time_struct {
439 	local64_t	time;
440 };
441 typedef struct rb_time_struct rb_time_t;
442 
443 #define MAX_NEST	5
444 
445 /*
446  * head_page == tail_page && head == tail then buffer is empty.
447  */
448 struct ring_buffer_per_cpu {
449 	int				cpu;
450 	atomic_t			record_disabled;
451 	atomic_t			resize_disabled;
452 	struct trace_buffer	*buffer;
453 	raw_spinlock_t			reader_lock;	/* serialize readers */
454 	arch_spinlock_t			lock;
455 	struct lock_class_key		lock_key;
456 	struct buffer_data_page		*free_page;
457 	unsigned long			nr_pages;
458 	unsigned int			current_context;
459 	struct list_head		*pages;
460 	struct buffer_page		*head_page;	/* read from head */
461 	struct buffer_page		*tail_page;	/* write to tail */
462 	struct buffer_page		*commit_page;	/* committed pages */
463 	struct buffer_page		*reader_page;
464 	unsigned long			lost_events;
465 	unsigned long			last_overrun;
466 	unsigned long			nest;
467 	local_t				entries_bytes;
468 	local_t				entries;
469 	local_t				overrun;
470 	local_t				commit_overrun;
471 	local_t				dropped_events;
472 	local_t				committing;
473 	local_t				commits;
474 	local_t				pages_touched;
475 	local_t				pages_lost;
476 	local_t				pages_read;
477 	long				last_pages_touch;
478 	size_t				shortest_full;
479 	unsigned long			read;
480 	unsigned long			read_bytes;
481 	rb_time_t			write_stamp;
482 	rb_time_t			before_stamp;
483 	u64				event_stamp[MAX_NEST];
484 	u64				read_stamp;
485 	/* pages removed since last reset */
486 	unsigned long			pages_removed;
487 	/* ring buffer pages to update, > 0 to add, < 0 to remove */
488 	long				nr_pages_to_update;
489 	struct list_head		new_pages; /* new pages to add */
490 	struct work_struct		update_pages_work;
491 	struct completion		update_done;
492 
493 	struct rb_irq_work		irq_work;
494 };
495 
496 struct trace_buffer {
497 	unsigned			flags;
498 	int				cpus;
499 	atomic_t			record_disabled;
500 	atomic_t			resizing;
501 	cpumask_var_t			cpumask;
502 
503 	struct lock_class_key		*reader_lock_key;
504 
505 	struct mutex			mutex;
506 
507 	struct ring_buffer_per_cpu	**buffers;
508 
509 	struct hlist_node		node;
510 	u64				(*clock)(void);
511 
512 	struct rb_irq_work		irq_work;
513 	bool				time_stamp_abs;
514 
515 	unsigned int			subbuf_size;
516 	unsigned int			subbuf_order;
517 	unsigned int			max_data_size;
518 };
519 
520 struct ring_buffer_iter {
521 	struct ring_buffer_per_cpu	*cpu_buffer;
522 	unsigned long			head;
523 	unsigned long			next_event;
524 	struct buffer_page		*head_page;
525 	struct buffer_page		*cache_reader_page;
526 	unsigned long			cache_read;
527 	unsigned long			cache_pages_removed;
528 	u64				read_stamp;
529 	u64				page_stamp;
530 	struct ring_buffer_event	*event;
531 	size_t				event_size;
532 	int				missed_events;
533 };
534 
535 int ring_buffer_print_page_header(struct trace_buffer *buffer, struct trace_seq *s)
536 {
537 	struct buffer_data_page field;
538 
539 	trace_seq_printf(s, "\tfield: u64 timestamp;\t"
540 			 "offset:0;\tsize:%u;\tsigned:%u;\n",
541 			 (unsigned int)sizeof(field.time_stamp),
542 			 (unsigned int)is_signed_type(u64));
543 
544 	trace_seq_printf(s, "\tfield: local_t commit;\t"
545 			 "offset:%u;\tsize:%u;\tsigned:%u;\n",
546 			 (unsigned int)offsetof(typeof(field), commit),
547 			 (unsigned int)sizeof(field.commit),
548 			 (unsigned int)is_signed_type(long));
549 
550 	trace_seq_printf(s, "\tfield: int overwrite;\t"
551 			 "offset:%u;\tsize:%u;\tsigned:%u;\n",
552 			 (unsigned int)offsetof(typeof(field), commit),
553 			 1,
554 			 (unsigned int)is_signed_type(long));
555 
556 	trace_seq_printf(s, "\tfield: char data;\t"
557 			 "offset:%u;\tsize:%u;\tsigned:%u;\n",
558 			 (unsigned int)offsetof(typeof(field), data),
559 			 (unsigned int)buffer->subbuf_size,
560 			 (unsigned int)is_signed_type(char));
561 
562 	return !trace_seq_has_overflowed(s);
563 }
564 
565 static inline void rb_time_read(rb_time_t *t, u64 *ret)
566 {
567 	*ret = local64_read(&t->time);
568 }
569 static void rb_time_set(rb_time_t *t, u64 val)
570 {
571 	local64_set(&t->time, val);
572 }
573 
574 /*
575  * Enable this to make sure that the event passed to
576  * ring_buffer_event_time_stamp() is not committed and also
577  * is on the buffer that it passed in.
578  */
579 //#define RB_VERIFY_EVENT
580 #ifdef RB_VERIFY_EVENT
581 static struct list_head *rb_list_head(struct list_head *list);
582 static void verify_event(struct ring_buffer_per_cpu *cpu_buffer,
583 			 void *event)
584 {
585 	struct buffer_page *page = cpu_buffer->commit_page;
586 	struct buffer_page *tail_page = READ_ONCE(cpu_buffer->tail_page);
587 	struct list_head *next;
588 	long commit, write;
589 	unsigned long addr = (unsigned long)event;
590 	bool done = false;
591 	int stop = 0;
592 
593 	/* Make sure the event exists and is not committed yet */
594 	do {
595 		if (page == tail_page || WARN_ON_ONCE(stop++ > 100))
596 			done = true;
597 		commit = local_read(&page->page->commit);
598 		write = local_read(&page->write);
599 		if (addr >= (unsigned long)&page->page->data[commit] &&
600 		    addr < (unsigned long)&page->page->data[write])
601 			return;
602 
603 		next = rb_list_head(page->list.next);
604 		page = list_entry(next, struct buffer_page, list);
605 	} while (!done);
606 	WARN_ON_ONCE(1);
607 }
608 #else
609 static inline void verify_event(struct ring_buffer_per_cpu *cpu_buffer,
610 			 void *event)
611 {
612 }
613 #endif
614 
615 /*
616  * The absolute time stamp drops the 5 MSBs and some clocks may
617  * require them. The rb_fix_abs_ts() will take a previous full
618  * time stamp, and add the 5 MSB of that time stamp on to the
619  * saved absolute time stamp. Then they are compared in case of
620  * the unlikely event that the latest time stamp incremented
621  * the 5 MSB.
622  */
623 static inline u64 rb_fix_abs_ts(u64 abs, u64 save_ts)
624 {
625 	if (save_ts & TS_MSB) {
626 		abs |= save_ts & TS_MSB;
627 		/* Check for overflow */
628 		if (unlikely(abs < save_ts))
629 			abs += 1ULL << 59;
630 	}
631 	return abs;
632 }
633 
634 static inline u64 rb_time_stamp(struct trace_buffer *buffer);
635 
636 /**
637  * ring_buffer_event_time_stamp - return the event's current time stamp
638  * @buffer: The buffer that the event is on
639  * @event: the event to get the time stamp of
640  *
641  * Note, this must be called after @event is reserved, and before it is
642  * committed to the ring buffer. And must be called from the same
643  * context where the event was reserved (normal, softirq, irq, etc).
644  *
645  * Returns the time stamp associated with the current event.
646  * If the event has an extended time stamp, then that is used as
647  * the time stamp to return.
648  * In the highly unlikely case that the event was nested more than
649  * the max nesting, then the write_stamp of the buffer is returned,
650  * otherwise  current time is returned, but that really neither of
651  * the last two cases should ever happen.
652  */
653 u64 ring_buffer_event_time_stamp(struct trace_buffer *buffer,
654 				 struct ring_buffer_event *event)
655 {
656 	struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[smp_processor_id()];
657 	unsigned int nest;
658 	u64 ts;
659 
660 	/* If the event includes an absolute time, then just use that */
661 	if (event->type_len == RINGBUF_TYPE_TIME_STAMP) {
662 		ts = rb_event_time_stamp(event);
663 		return rb_fix_abs_ts(ts, cpu_buffer->tail_page->page->time_stamp);
664 	}
665 
666 	nest = local_read(&cpu_buffer->committing);
667 	verify_event(cpu_buffer, event);
668 	if (WARN_ON_ONCE(!nest))
669 		goto fail;
670 
671 	/* Read the current saved nesting level time stamp */
672 	if (likely(--nest < MAX_NEST))
673 		return cpu_buffer->event_stamp[nest];
674 
675 	/* Shouldn't happen, warn if it does */
676 	WARN_ONCE(1, "nest (%d) greater than max", nest);
677 
678  fail:
679 	rb_time_read(&cpu_buffer->write_stamp, &ts);
680 
681 	return ts;
682 }
683 
684 /**
685  * ring_buffer_nr_pages - get the number of buffer pages in the ring buffer
686  * @buffer: The ring_buffer to get the number of pages from
687  * @cpu: The cpu of the ring_buffer to get the number of pages from
688  *
689  * Returns the number of pages used by a per_cpu buffer of the ring buffer.
690  */
691 size_t ring_buffer_nr_pages(struct trace_buffer *buffer, int cpu)
692 {
693 	return buffer->buffers[cpu]->nr_pages;
694 }
695 
696 /**
697  * ring_buffer_nr_dirty_pages - get the number of used pages in the ring buffer
698  * @buffer: The ring_buffer to get the number of pages from
699  * @cpu: The cpu of the ring_buffer to get the number of pages from
700  *
701  * Returns the number of pages that have content in the ring buffer.
702  */
703 size_t ring_buffer_nr_dirty_pages(struct trace_buffer *buffer, int cpu)
704 {
705 	size_t read;
706 	size_t lost;
707 	size_t cnt;
708 
709 	read = local_read(&buffer->buffers[cpu]->pages_read);
710 	lost = local_read(&buffer->buffers[cpu]->pages_lost);
711 	cnt = local_read(&buffer->buffers[cpu]->pages_touched);
712 
713 	if (WARN_ON_ONCE(cnt < lost))
714 		return 0;
715 
716 	cnt -= lost;
717 
718 	/* The reader can read an empty page, but not more than that */
719 	if (cnt < read) {
720 		WARN_ON_ONCE(read > cnt + 1);
721 		return 0;
722 	}
723 
724 	return cnt - read;
725 }
726 
727 static __always_inline bool full_hit(struct trace_buffer *buffer, int cpu, int full)
728 {
729 	struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
730 	size_t nr_pages;
731 	size_t dirty;
732 
733 	nr_pages = cpu_buffer->nr_pages;
734 	if (!nr_pages || !full)
735 		return true;
736 
737 	/*
738 	 * Add one as dirty will never equal nr_pages, as the sub-buffer
739 	 * that the writer is on is not counted as dirty.
740 	 * This is needed if "buffer_percent" is set to 100.
741 	 */
742 	dirty = ring_buffer_nr_dirty_pages(buffer, cpu) + 1;
743 
744 	return (dirty * 100) >= (full * nr_pages);
745 }
746 
747 /*
748  * rb_wake_up_waiters - wake up tasks waiting for ring buffer input
749  *
750  * Schedules a delayed work to wake up any task that is blocked on the
751  * ring buffer waiters queue.
752  */
753 static void rb_wake_up_waiters(struct irq_work *work)
754 {
755 	struct rb_irq_work *rbwork = container_of(work, struct rb_irq_work, work);
756 
757 	/* For waiters waiting for the first wake up */
758 	(void)atomic_fetch_inc_release(&rbwork->seq);
759 
760 	wake_up_all(&rbwork->waiters);
761 	if (rbwork->full_waiters_pending || rbwork->wakeup_full) {
762 		/* Only cpu_buffer sets the above flags */
763 		struct ring_buffer_per_cpu *cpu_buffer =
764 			container_of(rbwork, struct ring_buffer_per_cpu, irq_work);
765 
766 		/* Called from interrupt context */
767 		raw_spin_lock(&cpu_buffer->reader_lock);
768 		rbwork->wakeup_full = false;
769 		rbwork->full_waiters_pending = false;
770 
771 		/* Waking up all waiters, they will reset the shortest full */
772 		cpu_buffer->shortest_full = 0;
773 		raw_spin_unlock(&cpu_buffer->reader_lock);
774 
775 		wake_up_all(&rbwork->full_waiters);
776 	}
777 }
778 
779 /**
780  * ring_buffer_wake_waiters - wake up any waiters on this ring buffer
781  * @buffer: The ring buffer to wake waiters on
782  * @cpu: The CPU buffer to wake waiters on
783  *
784  * In the case of a file that represents a ring buffer is closing,
785  * it is prudent to wake up any waiters that are on this.
786  */
787 void ring_buffer_wake_waiters(struct trace_buffer *buffer, int cpu)
788 {
789 	struct ring_buffer_per_cpu *cpu_buffer;
790 	struct rb_irq_work *rbwork;
791 
792 	if (!buffer)
793 		return;
794 
795 	if (cpu == RING_BUFFER_ALL_CPUS) {
796 
797 		/* Wake up individual ones too. One level recursion */
798 		for_each_buffer_cpu(buffer, cpu)
799 			ring_buffer_wake_waiters(buffer, cpu);
800 
801 		rbwork = &buffer->irq_work;
802 	} else {
803 		if (WARN_ON_ONCE(!buffer->buffers))
804 			return;
805 		if (WARN_ON_ONCE(cpu >= nr_cpu_ids))
806 			return;
807 
808 		cpu_buffer = buffer->buffers[cpu];
809 		/* The CPU buffer may not have been initialized yet */
810 		if (!cpu_buffer)
811 			return;
812 		rbwork = &cpu_buffer->irq_work;
813 	}
814 
815 	/* This can be called in any context */
816 	irq_work_queue(&rbwork->work);
817 }
818 
819 static bool rb_watermark_hit(struct trace_buffer *buffer, int cpu, int full)
820 {
821 	struct ring_buffer_per_cpu *cpu_buffer;
822 	bool ret = false;
823 
824 	/* Reads of all CPUs always waits for any data */
825 	if (cpu == RING_BUFFER_ALL_CPUS)
826 		return !ring_buffer_empty(buffer);
827 
828 	cpu_buffer = buffer->buffers[cpu];
829 
830 	if (!ring_buffer_empty_cpu(buffer, cpu)) {
831 		unsigned long flags;
832 		bool pagebusy;
833 
834 		if (!full)
835 			return true;
836 
837 		raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
838 		pagebusy = cpu_buffer->reader_page == cpu_buffer->commit_page;
839 		ret = !pagebusy && full_hit(buffer, cpu, full);
840 
841 		if (!ret && (!cpu_buffer->shortest_full ||
842 			     cpu_buffer->shortest_full > full)) {
843 		    cpu_buffer->shortest_full = full;
844 		}
845 		raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
846 	}
847 	return ret;
848 }
849 
850 static inline bool
851 rb_wait_cond(struct rb_irq_work *rbwork, struct trace_buffer *buffer,
852 	     int cpu, int full, ring_buffer_cond_fn cond, void *data)
853 {
854 	if (rb_watermark_hit(buffer, cpu, full))
855 		return true;
856 
857 	if (cond(data))
858 		return true;
859 
860 	/*
861 	 * The events can happen in critical sections where
862 	 * checking a work queue can cause deadlocks.
863 	 * After adding a task to the queue, this flag is set
864 	 * only to notify events to try to wake up the queue
865 	 * using irq_work.
866 	 *
867 	 * We don't clear it even if the buffer is no longer
868 	 * empty. The flag only causes the next event to run
869 	 * irq_work to do the work queue wake up. The worse
870 	 * that can happen if we race with !trace_empty() is that
871 	 * an event will cause an irq_work to try to wake up
872 	 * an empty queue.
873 	 *
874 	 * There's no reason to protect this flag either, as
875 	 * the work queue and irq_work logic will do the necessary
876 	 * synchronization for the wake ups. The only thing
877 	 * that is necessary is that the wake up happens after
878 	 * a task has been queued. It's OK for spurious wake ups.
879 	 */
880 	if (full)
881 		rbwork->full_waiters_pending = true;
882 	else
883 		rbwork->waiters_pending = true;
884 
885 	return false;
886 }
887 
888 struct rb_wait_data {
889 	struct rb_irq_work		*irq_work;
890 	int				seq;
891 };
892 
893 /*
894  * The default wait condition for ring_buffer_wait() is to just to exit the
895  * wait loop the first time it is woken up.
896  */
897 static bool rb_wait_once(void *data)
898 {
899 	struct rb_wait_data *rdata = data;
900 	struct rb_irq_work *rbwork = rdata->irq_work;
901 
902 	return atomic_read_acquire(&rbwork->seq) != rdata->seq;
903 }
904 
905 /**
906  * ring_buffer_wait - wait for input to the ring buffer
907  * @buffer: buffer to wait on
908  * @cpu: the cpu buffer to wait on
909  * @full: wait until the percentage of pages are available, if @cpu != RING_BUFFER_ALL_CPUS
910  * @cond: condition function to break out of wait (NULL to run once)
911  * @data: the data to pass to @cond.
912  *
913  * If @cpu == RING_BUFFER_ALL_CPUS then the task will wake up as soon
914  * as data is added to any of the @buffer's cpu buffers. Otherwise
915  * it will wait for data to be added to a specific cpu buffer.
916  */
917 int ring_buffer_wait(struct trace_buffer *buffer, int cpu, int full,
918 		     ring_buffer_cond_fn cond, void *data)
919 {
920 	struct ring_buffer_per_cpu *cpu_buffer;
921 	struct wait_queue_head *waitq;
922 	struct rb_irq_work *rbwork;
923 	struct rb_wait_data rdata;
924 	int ret = 0;
925 
926 	/*
927 	 * Depending on what the caller is waiting for, either any
928 	 * data in any cpu buffer, or a specific buffer, put the
929 	 * caller on the appropriate wait queue.
930 	 */
931 	if (cpu == RING_BUFFER_ALL_CPUS) {
932 		rbwork = &buffer->irq_work;
933 		/* Full only makes sense on per cpu reads */
934 		full = 0;
935 	} else {
936 		if (!cpumask_test_cpu(cpu, buffer->cpumask))
937 			return -ENODEV;
938 		cpu_buffer = buffer->buffers[cpu];
939 		rbwork = &cpu_buffer->irq_work;
940 	}
941 
942 	if (full)
943 		waitq = &rbwork->full_waiters;
944 	else
945 		waitq = &rbwork->waiters;
946 
947 	/* Set up to exit loop as soon as it is woken */
948 	if (!cond) {
949 		cond = rb_wait_once;
950 		rdata.irq_work = rbwork;
951 		rdata.seq = atomic_read_acquire(&rbwork->seq);
952 		data = &rdata;
953 	}
954 
955 	ret = wait_event_interruptible((*waitq),
956 				rb_wait_cond(rbwork, buffer, cpu, full, cond, data));
957 
958 	return ret;
959 }
960 
961 /**
962  * ring_buffer_poll_wait - poll on buffer input
963  * @buffer: buffer to wait on
964  * @cpu: the cpu buffer to wait on
965  * @filp: the file descriptor
966  * @poll_table: The poll descriptor
967  * @full: wait until the percentage of pages are available, if @cpu != RING_BUFFER_ALL_CPUS
968  *
969  * If @cpu == RING_BUFFER_ALL_CPUS then the task will wake up as soon
970  * as data is added to any of the @buffer's cpu buffers. Otherwise
971  * it will wait for data to be added to a specific cpu buffer.
972  *
973  * Returns EPOLLIN | EPOLLRDNORM if data exists in the buffers,
974  * zero otherwise.
975  */
976 __poll_t ring_buffer_poll_wait(struct trace_buffer *buffer, int cpu,
977 			  struct file *filp, poll_table *poll_table, int full)
978 {
979 	struct ring_buffer_per_cpu *cpu_buffer;
980 	struct rb_irq_work *rbwork;
981 
982 	if (cpu == RING_BUFFER_ALL_CPUS) {
983 		rbwork = &buffer->irq_work;
984 		full = 0;
985 	} else {
986 		if (!cpumask_test_cpu(cpu, buffer->cpumask))
987 			return EPOLLERR;
988 
989 		cpu_buffer = buffer->buffers[cpu];
990 		rbwork = &cpu_buffer->irq_work;
991 	}
992 
993 	if (full) {
994 		poll_wait(filp, &rbwork->full_waiters, poll_table);
995 
996 		if (rb_watermark_hit(buffer, cpu, full))
997 			return EPOLLIN | EPOLLRDNORM;
998 		/*
999 		 * Only allow full_waiters_pending update to be seen after
1000 		 * the shortest_full is set (in rb_watermark_hit). If the
1001 		 * writer sees the full_waiters_pending flag set, it will
1002 		 * compare the amount in the ring buffer to shortest_full.
1003 		 * If the amount in the ring buffer is greater than the
1004 		 * shortest_full percent, it will call the irq_work handler
1005 		 * to wake up this list. The irq_handler will reset shortest_full
1006 		 * back to zero. That's done under the reader_lock, but
1007 		 * the below smp_mb() makes sure that the update to
1008 		 * full_waiters_pending doesn't leak up into the above.
1009 		 */
1010 		smp_mb();
1011 		rbwork->full_waiters_pending = true;
1012 		return 0;
1013 	}
1014 
1015 	poll_wait(filp, &rbwork->waiters, poll_table);
1016 	rbwork->waiters_pending = true;
1017 
1018 	/*
1019 	 * There's a tight race between setting the waiters_pending and
1020 	 * checking if the ring buffer is empty.  Once the waiters_pending bit
1021 	 * is set, the next event will wake the task up, but we can get stuck
1022 	 * if there's only a single event in.
1023 	 *
1024 	 * FIXME: Ideally, we need a memory barrier on the writer side as well,
1025 	 * but adding a memory barrier to all events will cause too much of a
1026 	 * performance hit in the fast path.  We only need a memory barrier when
1027 	 * the buffer goes from empty to having content.  But as this race is
1028 	 * extremely small, and it's not a problem if another event comes in, we
1029 	 * will fix it later.
1030 	 */
1031 	smp_mb();
1032 
1033 	if ((cpu == RING_BUFFER_ALL_CPUS && !ring_buffer_empty(buffer)) ||
1034 	    (cpu != RING_BUFFER_ALL_CPUS && !ring_buffer_empty_cpu(buffer, cpu)))
1035 		return EPOLLIN | EPOLLRDNORM;
1036 	return 0;
1037 }
1038 
1039 /* buffer may be either ring_buffer or ring_buffer_per_cpu */
1040 #define RB_WARN_ON(b, cond)						\
1041 	({								\
1042 		int _____ret = unlikely(cond);				\
1043 		if (_____ret) {						\
1044 			if (__same_type(*(b), struct ring_buffer_per_cpu)) { \
1045 				struct ring_buffer_per_cpu *__b =	\
1046 					(void *)b;			\
1047 				atomic_inc(&__b->buffer->record_disabled); \
1048 			} else						\
1049 				atomic_inc(&b->record_disabled);	\
1050 			WARN_ON(1);					\
1051 		}							\
1052 		_____ret;						\
1053 	})
1054 
1055 /* Up this if you want to test the TIME_EXTENTS and normalization */
1056 #define DEBUG_SHIFT 0
1057 
1058 static inline u64 rb_time_stamp(struct trace_buffer *buffer)
1059 {
1060 	u64 ts;
1061 
1062 	/* Skip retpolines :-( */
1063 	if (IS_ENABLED(CONFIG_MITIGATION_RETPOLINE) && likely(buffer->clock == trace_clock_local))
1064 		ts = trace_clock_local();
1065 	else
1066 		ts = buffer->clock();
1067 
1068 	/* shift to debug/test normalization and TIME_EXTENTS */
1069 	return ts << DEBUG_SHIFT;
1070 }
1071 
1072 u64 ring_buffer_time_stamp(struct trace_buffer *buffer)
1073 {
1074 	u64 time;
1075 
1076 	preempt_disable_notrace();
1077 	time = rb_time_stamp(buffer);
1078 	preempt_enable_notrace();
1079 
1080 	return time;
1081 }
1082 EXPORT_SYMBOL_GPL(ring_buffer_time_stamp);
1083 
1084 void ring_buffer_normalize_time_stamp(struct trace_buffer *buffer,
1085 				      int cpu, u64 *ts)
1086 {
1087 	/* Just stupid testing the normalize function and deltas */
1088 	*ts >>= DEBUG_SHIFT;
1089 }
1090 EXPORT_SYMBOL_GPL(ring_buffer_normalize_time_stamp);
1091 
1092 /*
1093  * Making the ring buffer lockless makes things tricky.
1094  * Although writes only happen on the CPU that they are on,
1095  * and they only need to worry about interrupts. Reads can
1096  * happen on any CPU.
1097  *
1098  * The reader page is always off the ring buffer, but when the
1099  * reader finishes with a page, it needs to swap its page with
1100  * a new one from the buffer. The reader needs to take from
1101  * the head (writes go to the tail). But if a writer is in overwrite
1102  * mode and wraps, it must push the head page forward.
1103  *
1104  * Here lies the problem.
1105  *
1106  * The reader must be careful to replace only the head page, and
1107  * not another one. As described at the top of the file in the
1108  * ASCII art, the reader sets its old page to point to the next
1109  * page after head. It then sets the page after head to point to
1110  * the old reader page. But if the writer moves the head page
1111  * during this operation, the reader could end up with the tail.
1112  *
1113  * We use cmpxchg to help prevent this race. We also do something
1114  * special with the page before head. We set the LSB to 1.
1115  *
1116  * When the writer must push the page forward, it will clear the
1117  * bit that points to the head page, move the head, and then set
1118  * the bit that points to the new head page.
1119  *
1120  * We also don't want an interrupt coming in and moving the head
1121  * page on another writer. Thus we use the second LSB to catch
1122  * that too. Thus:
1123  *
1124  * head->list->prev->next        bit 1          bit 0
1125  *                              -------        -------
1126  * Normal page                     0              0
1127  * Points to head page             0              1
1128  * New head page                   1              0
1129  *
1130  * Note we can not trust the prev pointer of the head page, because:
1131  *
1132  * +----+       +-----+        +-----+
1133  * |    |------>|  T  |---X--->|  N  |
1134  * |    |<------|     |        |     |
1135  * +----+       +-----+        +-----+
1136  *   ^                           ^ |
1137  *   |          +-----+          | |
1138  *   +----------|  R  |----------+ |
1139  *              |     |<-----------+
1140  *              +-----+
1141  *
1142  * Key:  ---X-->  HEAD flag set in pointer
1143  *         T      Tail page
1144  *         R      Reader page
1145  *         N      Next page
1146  *
1147  * (see __rb_reserve_next() to see where this happens)
1148  *
1149  *  What the above shows is that the reader just swapped out
1150  *  the reader page with a page in the buffer, but before it
1151  *  could make the new header point back to the new page added
1152  *  it was preempted by a writer. The writer moved forward onto
1153  *  the new page added by the reader and is about to move forward
1154  *  again.
1155  *
1156  *  You can see, it is legitimate for the previous pointer of
1157  *  the head (or any page) not to point back to itself. But only
1158  *  temporarily.
1159  */
1160 
1161 #define RB_PAGE_NORMAL		0UL
1162 #define RB_PAGE_HEAD		1UL
1163 #define RB_PAGE_UPDATE		2UL
1164 
1165 
1166 #define RB_FLAG_MASK		3UL
1167 
1168 /* PAGE_MOVED is not part of the mask */
1169 #define RB_PAGE_MOVED		4UL
1170 
1171 /*
1172  * rb_list_head - remove any bit
1173  */
1174 static struct list_head *rb_list_head(struct list_head *list)
1175 {
1176 	unsigned long val = (unsigned long)list;
1177 
1178 	return (struct list_head *)(val & ~RB_FLAG_MASK);
1179 }
1180 
1181 /*
1182  * rb_is_head_page - test if the given page is the head page
1183  *
1184  * Because the reader may move the head_page pointer, we can
1185  * not trust what the head page is (it may be pointing to
1186  * the reader page). But if the next page is a header page,
1187  * its flags will be non zero.
1188  */
1189 static inline int
1190 rb_is_head_page(struct buffer_page *page, struct list_head *list)
1191 {
1192 	unsigned long val;
1193 
1194 	val = (unsigned long)list->next;
1195 
1196 	if ((val & ~RB_FLAG_MASK) != (unsigned long)&page->list)
1197 		return RB_PAGE_MOVED;
1198 
1199 	return val & RB_FLAG_MASK;
1200 }
1201 
1202 /*
1203  * rb_is_reader_page
1204  *
1205  * The unique thing about the reader page, is that, if the
1206  * writer is ever on it, the previous pointer never points
1207  * back to the reader page.
1208  */
1209 static bool rb_is_reader_page(struct buffer_page *page)
1210 {
1211 	struct list_head *list = page->list.prev;
1212 
1213 	return rb_list_head(list->next) != &page->list;
1214 }
1215 
1216 /*
1217  * rb_set_list_to_head - set a list_head to be pointing to head.
1218  */
1219 static void rb_set_list_to_head(struct list_head *list)
1220 {
1221 	unsigned long *ptr;
1222 
1223 	ptr = (unsigned long *)&list->next;
1224 	*ptr |= RB_PAGE_HEAD;
1225 	*ptr &= ~RB_PAGE_UPDATE;
1226 }
1227 
1228 /*
1229  * rb_head_page_activate - sets up head page
1230  */
1231 static void rb_head_page_activate(struct ring_buffer_per_cpu *cpu_buffer)
1232 {
1233 	struct buffer_page *head;
1234 
1235 	head = cpu_buffer->head_page;
1236 	if (!head)
1237 		return;
1238 
1239 	/*
1240 	 * Set the previous list pointer to have the HEAD flag.
1241 	 */
1242 	rb_set_list_to_head(head->list.prev);
1243 }
1244 
1245 static void rb_list_head_clear(struct list_head *list)
1246 {
1247 	unsigned long *ptr = (unsigned long *)&list->next;
1248 
1249 	*ptr &= ~RB_FLAG_MASK;
1250 }
1251 
1252 /*
1253  * rb_head_page_deactivate - clears head page ptr (for free list)
1254  */
1255 static void
1256 rb_head_page_deactivate(struct ring_buffer_per_cpu *cpu_buffer)
1257 {
1258 	struct list_head *hd;
1259 
1260 	/* Go through the whole list and clear any pointers found. */
1261 	rb_list_head_clear(cpu_buffer->pages);
1262 
1263 	list_for_each(hd, cpu_buffer->pages)
1264 		rb_list_head_clear(hd);
1265 }
1266 
1267 static int rb_head_page_set(struct ring_buffer_per_cpu *cpu_buffer,
1268 			    struct buffer_page *head,
1269 			    struct buffer_page *prev,
1270 			    int old_flag, int new_flag)
1271 {
1272 	struct list_head *list;
1273 	unsigned long val = (unsigned long)&head->list;
1274 	unsigned long ret;
1275 
1276 	list = &prev->list;
1277 
1278 	val &= ~RB_FLAG_MASK;
1279 
1280 	ret = cmpxchg((unsigned long *)&list->next,
1281 		      val | old_flag, val | new_flag);
1282 
1283 	/* check if the reader took the page */
1284 	if ((ret & ~RB_FLAG_MASK) != val)
1285 		return RB_PAGE_MOVED;
1286 
1287 	return ret & RB_FLAG_MASK;
1288 }
1289 
1290 static int rb_head_page_set_update(struct ring_buffer_per_cpu *cpu_buffer,
1291 				   struct buffer_page *head,
1292 				   struct buffer_page *prev,
1293 				   int old_flag)
1294 {
1295 	return rb_head_page_set(cpu_buffer, head, prev,
1296 				old_flag, RB_PAGE_UPDATE);
1297 }
1298 
1299 static int rb_head_page_set_head(struct ring_buffer_per_cpu *cpu_buffer,
1300 				 struct buffer_page *head,
1301 				 struct buffer_page *prev,
1302 				 int old_flag)
1303 {
1304 	return rb_head_page_set(cpu_buffer, head, prev,
1305 				old_flag, RB_PAGE_HEAD);
1306 }
1307 
1308 static int rb_head_page_set_normal(struct ring_buffer_per_cpu *cpu_buffer,
1309 				   struct buffer_page *head,
1310 				   struct buffer_page *prev,
1311 				   int old_flag)
1312 {
1313 	return rb_head_page_set(cpu_buffer, head, prev,
1314 				old_flag, RB_PAGE_NORMAL);
1315 }
1316 
1317 static inline void rb_inc_page(struct buffer_page **bpage)
1318 {
1319 	struct list_head *p = rb_list_head((*bpage)->list.next);
1320 
1321 	*bpage = list_entry(p, struct buffer_page, list);
1322 }
1323 
1324 static struct buffer_page *
1325 rb_set_head_page(struct ring_buffer_per_cpu *cpu_buffer)
1326 {
1327 	struct buffer_page *head;
1328 	struct buffer_page *page;
1329 	struct list_head *list;
1330 	int i;
1331 
1332 	if (RB_WARN_ON(cpu_buffer, !cpu_buffer->head_page))
1333 		return NULL;
1334 
1335 	/* sanity check */
1336 	list = cpu_buffer->pages;
1337 	if (RB_WARN_ON(cpu_buffer, rb_list_head(list->prev->next) != list))
1338 		return NULL;
1339 
1340 	page = head = cpu_buffer->head_page;
1341 	/*
1342 	 * It is possible that the writer moves the header behind
1343 	 * where we started, and we miss in one loop.
1344 	 * A second loop should grab the header, but we'll do
1345 	 * three loops just because I'm paranoid.
1346 	 */
1347 	for (i = 0; i < 3; i++) {
1348 		do {
1349 			if (rb_is_head_page(page, page->list.prev)) {
1350 				cpu_buffer->head_page = page;
1351 				return page;
1352 			}
1353 			rb_inc_page(&page);
1354 		} while (page != head);
1355 	}
1356 
1357 	RB_WARN_ON(cpu_buffer, 1);
1358 
1359 	return NULL;
1360 }
1361 
1362 static bool rb_head_page_replace(struct buffer_page *old,
1363 				struct buffer_page *new)
1364 {
1365 	unsigned long *ptr = (unsigned long *)&old->list.prev->next;
1366 	unsigned long val;
1367 
1368 	val = *ptr & ~RB_FLAG_MASK;
1369 	val |= RB_PAGE_HEAD;
1370 
1371 	return try_cmpxchg(ptr, &val, (unsigned long)&new->list);
1372 }
1373 
1374 /*
1375  * rb_tail_page_update - move the tail page forward
1376  */
1377 static void rb_tail_page_update(struct ring_buffer_per_cpu *cpu_buffer,
1378 			       struct buffer_page *tail_page,
1379 			       struct buffer_page *next_page)
1380 {
1381 	unsigned long old_entries;
1382 	unsigned long old_write;
1383 
1384 	/*
1385 	 * The tail page now needs to be moved forward.
1386 	 *
1387 	 * We need to reset the tail page, but without messing
1388 	 * with possible erasing of data brought in by interrupts
1389 	 * that have moved the tail page and are currently on it.
1390 	 *
1391 	 * We add a counter to the write field to denote this.
1392 	 */
1393 	old_write = local_add_return(RB_WRITE_INTCNT, &next_page->write);
1394 	old_entries = local_add_return(RB_WRITE_INTCNT, &next_page->entries);
1395 
1396 	/*
1397 	 * Just make sure we have seen our old_write and synchronize
1398 	 * with any interrupts that come in.
1399 	 */
1400 	barrier();
1401 
1402 	/*
1403 	 * If the tail page is still the same as what we think
1404 	 * it is, then it is up to us to update the tail
1405 	 * pointer.
1406 	 */
1407 	if (tail_page == READ_ONCE(cpu_buffer->tail_page)) {
1408 		/* Zero the write counter */
1409 		unsigned long val = old_write & ~RB_WRITE_MASK;
1410 		unsigned long eval = old_entries & ~RB_WRITE_MASK;
1411 
1412 		/*
1413 		 * This will only succeed if an interrupt did
1414 		 * not come in and change it. In which case, we
1415 		 * do not want to modify it.
1416 		 *
1417 		 * We add (void) to let the compiler know that we do not care
1418 		 * about the return value of these functions. We use the
1419 		 * cmpxchg to only update if an interrupt did not already
1420 		 * do it for us. If the cmpxchg fails, we don't care.
1421 		 */
1422 		(void)local_cmpxchg(&next_page->write, old_write, val);
1423 		(void)local_cmpxchg(&next_page->entries, old_entries, eval);
1424 
1425 		/*
1426 		 * No need to worry about races with clearing out the commit.
1427 		 * it only can increment when a commit takes place. But that
1428 		 * only happens in the outer most nested commit.
1429 		 */
1430 		local_set(&next_page->page->commit, 0);
1431 
1432 		/* Either we update tail_page or an interrupt does */
1433 		if (try_cmpxchg(&cpu_buffer->tail_page, &tail_page, next_page))
1434 			local_inc(&cpu_buffer->pages_touched);
1435 	}
1436 }
1437 
1438 static void rb_check_bpage(struct ring_buffer_per_cpu *cpu_buffer,
1439 			  struct buffer_page *bpage)
1440 {
1441 	unsigned long val = (unsigned long)bpage;
1442 
1443 	RB_WARN_ON(cpu_buffer, val & RB_FLAG_MASK);
1444 }
1445 
1446 /**
1447  * rb_check_pages - integrity check of buffer pages
1448  * @cpu_buffer: CPU buffer with pages to test
1449  *
1450  * As a safety measure we check to make sure the data pages have not
1451  * been corrupted.
1452  */
1453 static void rb_check_pages(struct ring_buffer_per_cpu *cpu_buffer)
1454 {
1455 	struct list_head *head = rb_list_head(cpu_buffer->pages);
1456 	struct list_head *tmp;
1457 
1458 	if (RB_WARN_ON(cpu_buffer,
1459 			rb_list_head(rb_list_head(head->next)->prev) != head))
1460 		return;
1461 
1462 	if (RB_WARN_ON(cpu_buffer,
1463 			rb_list_head(rb_list_head(head->prev)->next) != head))
1464 		return;
1465 
1466 	for (tmp = rb_list_head(head->next); tmp != head; tmp = rb_list_head(tmp->next)) {
1467 		if (RB_WARN_ON(cpu_buffer,
1468 				rb_list_head(rb_list_head(tmp->next)->prev) != tmp))
1469 			return;
1470 
1471 		if (RB_WARN_ON(cpu_buffer,
1472 				rb_list_head(rb_list_head(tmp->prev)->next) != tmp))
1473 			return;
1474 	}
1475 }
1476 
1477 static int __rb_allocate_pages(struct ring_buffer_per_cpu *cpu_buffer,
1478 		long nr_pages, struct list_head *pages)
1479 {
1480 	struct buffer_page *bpage, *tmp;
1481 	bool user_thread = current->mm != NULL;
1482 	gfp_t mflags;
1483 	long i;
1484 
1485 	/*
1486 	 * Check if the available memory is there first.
1487 	 * Note, si_mem_available() only gives us a rough estimate of available
1488 	 * memory. It may not be accurate. But we don't care, we just want
1489 	 * to prevent doing any allocation when it is obvious that it is
1490 	 * not going to succeed.
1491 	 */
1492 	i = si_mem_available();
1493 	if (i < nr_pages)
1494 		return -ENOMEM;
1495 
1496 	/*
1497 	 * __GFP_RETRY_MAYFAIL flag makes sure that the allocation fails
1498 	 * gracefully without invoking oom-killer and the system is not
1499 	 * destabilized.
1500 	 */
1501 	mflags = GFP_KERNEL | __GFP_RETRY_MAYFAIL;
1502 
1503 	/*
1504 	 * If a user thread allocates too much, and si_mem_available()
1505 	 * reports there's enough memory, even though there is not.
1506 	 * Make sure the OOM killer kills this thread. This can happen
1507 	 * even with RETRY_MAYFAIL because another task may be doing
1508 	 * an allocation after this task has taken all memory.
1509 	 * This is the task the OOM killer needs to take out during this
1510 	 * loop, even if it was triggered by an allocation somewhere else.
1511 	 */
1512 	if (user_thread)
1513 		set_current_oom_origin();
1514 	for (i = 0; i < nr_pages; i++) {
1515 		struct page *page;
1516 
1517 		bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),
1518 				    mflags, cpu_to_node(cpu_buffer->cpu));
1519 		if (!bpage)
1520 			goto free_pages;
1521 
1522 		rb_check_bpage(cpu_buffer, bpage);
1523 
1524 		list_add(&bpage->list, pages);
1525 
1526 		page = alloc_pages_node(cpu_to_node(cpu_buffer->cpu),
1527 					mflags | __GFP_ZERO,
1528 					cpu_buffer->buffer->subbuf_order);
1529 		if (!page)
1530 			goto free_pages;
1531 		bpage->page = page_address(page);
1532 		bpage->order = cpu_buffer->buffer->subbuf_order;
1533 		rb_init_page(bpage->page);
1534 
1535 		if (user_thread && fatal_signal_pending(current))
1536 			goto free_pages;
1537 	}
1538 	if (user_thread)
1539 		clear_current_oom_origin();
1540 
1541 	return 0;
1542 
1543 free_pages:
1544 	list_for_each_entry_safe(bpage, tmp, pages, list) {
1545 		list_del_init(&bpage->list);
1546 		free_buffer_page(bpage);
1547 	}
1548 	if (user_thread)
1549 		clear_current_oom_origin();
1550 
1551 	return -ENOMEM;
1552 }
1553 
1554 static int rb_allocate_pages(struct ring_buffer_per_cpu *cpu_buffer,
1555 			     unsigned long nr_pages)
1556 {
1557 	LIST_HEAD(pages);
1558 
1559 	WARN_ON(!nr_pages);
1560 
1561 	if (__rb_allocate_pages(cpu_buffer, nr_pages, &pages))
1562 		return -ENOMEM;
1563 
1564 	/*
1565 	 * The ring buffer page list is a circular list that does not
1566 	 * start and end with a list head. All page list items point to
1567 	 * other pages.
1568 	 */
1569 	cpu_buffer->pages = pages.next;
1570 	list_del(&pages);
1571 
1572 	cpu_buffer->nr_pages = nr_pages;
1573 
1574 	rb_check_pages(cpu_buffer);
1575 
1576 	return 0;
1577 }
1578 
1579 static struct ring_buffer_per_cpu *
1580 rb_allocate_cpu_buffer(struct trace_buffer *buffer, long nr_pages, int cpu)
1581 {
1582 	struct ring_buffer_per_cpu *cpu_buffer;
1583 	struct buffer_page *bpage;
1584 	struct page *page;
1585 	int ret;
1586 
1587 	cpu_buffer = kzalloc_node(ALIGN(sizeof(*cpu_buffer), cache_line_size()),
1588 				  GFP_KERNEL, cpu_to_node(cpu));
1589 	if (!cpu_buffer)
1590 		return NULL;
1591 
1592 	cpu_buffer->cpu = cpu;
1593 	cpu_buffer->buffer = buffer;
1594 	raw_spin_lock_init(&cpu_buffer->reader_lock);
1595 	lockdep_set_class(&cpu_buffer->reader_lock, buffer->reader_lock_key);
1596 	cpu_buffer->lock = (arch_spinlock_t)__ARCH_SPIN_LOCK_UNLOCKED;
1597 	INIT_WORK(&cpu_buffer->update_pages_work, update_pages_handler);
1598 	init_completion(&cpu_buffer->update_done);
1599 	init_irq_work(&cpu_buffer->irq_work.work, rb_wake_up_waiters);
1600 	init_waitqueue_head(&cpu_buffer->irq_work.waiters);
1601 	init_waitqueue_head(&cpu_buffer->irq_work.full_waiters);
1602 
1603 	bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),
1604 			    GFP_KERNEL, cpu_to_node(cpu));
1605 	if (!bpage)
1606 		goto fail_free_buffer;
1607 
1608 	rb_check_bpage(cpu_buffer, bpage);
1609 
1610 	cpu_buffer->reader_page = bpage;
1611 
1612 	page = alloc_pages_node(cpu_to_node(cpu), GFP_KERNEL | __GFP_ZERO,
1613 				cpu_buffer->buffer->subbuf_order);
1614 	if (!page)
1615 		goto fail_free_reader;
1616 	bpage->page = page_address(page);
1617 	rb_init_page(bpage->page);
1618 
1619 	INIT_LIST_HEAD(&cpu_buffer->reader_page->list);
1620 	INIT_LIST_HEAD(&cpu_buffer->new_pages);
1621 
1622 	ret = rb_allocate_pages(cpu_buffer, nr_pages);
1623 	if (ret < 0)
1624 		goto fail_free_reader;
1625 
1626 	cpu_buffer->head_page
1627 		= list_entry(cpu_buffer->pages, struct buffer_page, list);
1628 	cpu_buffer->tail_page = cpu_buffer->commit_page = cpu_buffer->head_page;
1629 
1630 	rb_head_page_activate(cpu_buffer);
1631 
1632 	return cpu_buffer;
1633 
1634  fail_free_reader:
1635 	free_buffer_page(cpu_buffer->reader_page);
1636 
1637  fail_free_buffer:
1638 	kfree(cpu_buffer);
1639 	return NULL;
1640 }
1641 
1642 static void rb_free_cpu_buffer(struct ring_buffer_per_cpu *cpu_buffer)
1643 {
1644 	struct list_head *head = cpu_buffer->pages;
1645 	struct buffer_page *bpage, *tmp;
1646 
1647 	irq_work_sync(&cpu_buffer->irq_work.work);
1648 
1649 	free_buffer_page(cpu_buffer->reader_page);
1650 
1651 	if (head) {
1652 		rb_head_page_deactivate(cpu_buffer);
1653 
1654 		list_for_each_entry_safe(bpage, tmp, head, list) {
1655 			list_del_init(&bpage->list);
1656 			free_buffer_page(bpage);
1657 		}
1658 		bpage = list_entry(head, struct buffer_page, list);
1659 		free_buffer_page(bpage);
1660 	}
1661 
1662 	free_page((unsigned long)cpu_buffer->free_page);
1663 
1664 	kfree(cpu_buffer);
1665 }
1666 
1667 /**
1668  * __ring_buffer_alloc - allocate a new ring_buffer
1669  * @size: the size in bytes per cpu that is needed.
1670  * @flags: attributes to set for the ring buffer.
1671  * @key: ring buffer reader_lock_key.
1672  *
1673  * Currently the only flag that is available is the RB_FL_OVERWRITE
1674  * flag. This flag means that the buffer will overwrite old data
1675  * when the buffer wraps. If this flag is not set, the buffer will
1676  * drop data when the tail hits the head.
1677  */
1678 struct trace_buffer *__ring_buffer_alloc(unsigned long size, unsigned flags,
1679 					struct lock_class_key *key)
1680 {
1681 	struct trace_buffer *buffer;
1682 	long nr_pages;
1683 	int bsize;
1684 	int cpu;
1685 	int ret;
1686 
1687 	/* keep it in its own cache line */
1688 	buffer = kzalloc(ALIGN(sizeof(*buffer), cache_line_size()),
1689 			 GFP_KERNEL);
1690 	if (!buffer)
1691 		return NULL;
1692 
1693 	if (!zalloc_cpumask_var(&buffer->cpumask, GFP_KERNEL))
1694 		goto fail_free_buffer;
1695 
1696 	/* Default buffer page size - one system page */
1697 	buffer->subbuf_order = 0;
1698 	buffer->subbuf_size = PAGE_SIZE - BUF_PAGE_HDR_SIZE;
1699 
1700 	/* Max payload is buffer page size - header (8bytes) */
1701 	buffer->max_data_size = buffer->subbuf_size - (sizeof(u32) * 2);
1702 
1703 	nr_pages = DIV_ROUND_UP(size, buffer->subbuf_size);
1704 	buffer->flags = flags;
1705 	buffer->clock = trace_clock_local;
1706 	buffer->reader_lock_key = key;
1707 
1708 	init_irq_work(&buffer->irq_work.work, rb_wake_up_waiters);
1709 	init_waitqueue_head(&buffer->irq_work.waiters);
1710 
1711 	/* need at least two pages */
1712 	if (nr_pages < 2)
1713 		nr_pages = 2;
1714 
1715 	buffer->cpus = nr_cpu_ids;
1716 
1717 	bsize = sizeof(void *) * nr_cpu_ids;
1718 	buffer->buffers = kzalloc(ALIGN(bsize, cache_line_size()),
1719 				  GFP_KERNEL);
1720 	if (!buffer->buffers)
1721 		goto fail_free_cpumask;
1722 
1723 	cpu = raw_smp_processor_id();
1724 	cpumask_set_cpu(cpu, buffer->cpumask);
1725 	buffer->buffers[cpu] = rb_allocate_cpu_buffer(buffer, nr_pages, cpu);
1726 	if (!buffer->buffers[cpu])
1727 		goto fail_free_buffers;
1728 
1729 	ret = cpuhp_state_add_instance(CPUHP_TRACE_RB_PREPARE, &buffer->node);
1730 	if (ret < 0)
1731 		goto fail_free_buffers;
1732 
1733 	mutex_init(&buffer->mutex);
1734 
1735 	return buffer;
1736 
1737  fail_free_buffers:
1738 	for_each_buffer_cpu(buffer, cpu) {
1739 		if (buffer->buffers[cpu])
1740 			rb_free_cpu_buffer(buffer->buffers[cpu]);
1741 	}
1742 	kfree(buffer->buffers);
1743 
1744  fail_free_cpumask:
1745 	free_cpumask_var(buffer->cpumask);
1746 
1747  fail_free_buffer:
1748 	kfree(buffer);
1749 	return NULL;
1750 }
1751 EXPORT_SYMBOL_GPL(__ring_buffer_alloc);
1752 
1753 /**
1754  * ring_buffer_free - free a ring buffer.
1755  * @buffer: the buffer to free.
1756  */
1757 void
1758 ring_buffer_free(struct trace_buffer *buffer)
1759 {
1760 	int cpu;
1761 
1762 	cpuhp_state_remove_instance(CPUHP_TRACE_RB_PREPARE, &buffer->node);
1763 
1764 	irq_work_sync(&buffer->irq_work.work);
1765 
1766 	for_each_buffer_cpu(buffer, cpu)
1767 		rb_free_cpu_buffer(buffer->buffers[cpu]);
1768 
1769 	kfree(buffer->buffers);
1770 	free_cpumask_var(buffer->cpumask);
1771 
1772 	kfree(buffer);
1773 }
1774 EXPORT_SYMBOL_GPL(ring_buffer_free);
1775 
1776 void ring_buffer_set_clock(struct trace_buffer *buffer,
1777 			   u64 (*clock)(void))
1778 {
1779 	buffer->clock = clock;
1780 }
1781 
1782 void ring_buffer_set_time_stamp_abs(struct trace_buffer *buffer, bool abs)
1783 {
1784 	buffer->time_stamp_abs = abs;
1785 }
1786 
1787 bool ring_buffer_time_stamp_abs(struct trace_buffer *buffer)
1788 {
1789 	return buffer->time_stamp_abs;
1790 }
1791 
1792 static void rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer);
1793 
1794 static inline unsigned long rb_page_entries(struct buffer_page *bpage)
1795 {
1796 	return local_read(&bpage->entries) & RB_WRITE_MASK;
1797 }
1798 
1799 static inline unsigned long rb_page_write(struct buffer_page *bpage)
1800 {
1801 	return local_read(&bpage->write) & RB_WRITE_MASK;
1802 }
1803 
1804 static bool
1805 rb_remove_pages(struct ring_buffer_per_cpu *cpu_buffer, unsigned long nr_pages)
1806 {
1807 	struct list_head *tail_page, *to_remove, *next_page;
1808 	struct buffer_page *to_remove_page, *tmp_iter_page;
1809 	struct buffer_page *last_page, *first_page;
1810 	unsigned long nr_removed;
1811 	unsigned long head_bit;
1812 	int page_entries;
1813 
1814 	head_bit = 0;
1815 
1816 	raw_spin_lock_irq(&cpu_buffer->reader_lock);
1817 	atomic_inc(&cpu_buffer->record_disabled);
1818 	/*
1819 	 * We don't race with the readers since we have acquired the reader
1820 	 * lock. We also don't race with writers after disabling recording.
1821 	 * This makes it easy to figure out the first and the last page to be
1822 	 * removed from the list. We unlink all the pages in between including
1823 	 * the first and last pages. This is done in a busy loop so that we
1824 	 * lose the least number of traces.
1825 	 * The pages are freed after we restart recording and unlock readers.
1826 	 */
1827 	tail_page = &cpu_buffer->tail_page->list;
1828 
1829 	/*
1830 	 * tail page might be on reader page, we remove the next page
1831 	 * from the ring buffer
1832 	 */
1833 	if (cpu_buffer->tail_page == cpu_buffer->reader_page)
1834 		tail_page = rb_list_head(tail_page->next);
1835 	to_remove = tail_page;
1836 
1837 	/* start of pages to remove */
1838 	first_page = list_entry(rb_list_head(to_remove->next),
1839 				struct buffer_page, list);
1840 
1841 	for (nr_removed = 0; nr_removed < nr_pages; nr_removed++) {
1842 		to_remove = rb_list_head(to_remove)->next;
1843 		head_bit |= (unsigned long)to_remove & RB_PAGE_HEAD;
1844 	}
1845 	/* Read iterators need to reset themselves when some pages removed */
1846 	cpu_buffer->pages_removed += nr_removed;
1847 
1848 	next_page = rb_list_head(to_remove)->next;
1849 
1850 	/*
1851 	 * Now we remove all pages between tail_page and next_page.
1852 	 * Make sure that we have head_bit value preserved for the
1853 	 * next page
1854 	 */
1855 	tail_page->next = (struct list_head *)((unsigned long)next_page |
1856 						head_bit);
1857 	next_page = rb_list_head(next_page);
1858 	next_page->prev = tail_page;
1859 
1860 	/* make sure pages points to a valid page in the ring buffer */
1861 	cpu_buffer->pages = next_page;
1862 
1863 	/* update head page */
1864 	if (head_bit)
1865 		cpu_buffer->head_page = list_entry(next_page,
1866 						struct buffer_page, list);
1867 
1868 	/* pages are removed, resume tracing and then free the pages */
1869 	atomic_dec(&cpu_buffer->record_disabled);
1870 	raw_spin_unlock_irq(&cpu_buffer->reader_lock);
1871 
1872 	RB_WARN_ON(cpu_buffer, list_empty(cpu_buffer->pages));
1873 
1874 	/* last buffer page to remove */
1875 	last_page = list_entry(rb_list_head(to_remove), struct buffer_page,
1876 				list);
1877 	tmp_iter_page = first_page;
1878 
1879 	do {
1880 		cond_resched();
1881 
1882 		to_remove_page = tmp_iter_page;
1883 		rb_inc_page(&tmp_iter_page);
1884 
1885 		/* update the counters */
1886 		page_entries = rb_page_entries(to_remove_page);
1887 		if (page_entries) {
1888 			/*
1889 			 * If something was added to this page, it was full
1890 			 * since it is not the tail page. So we deduct the
1891 			 * bytes consumed in ring buffer from here.
1892 			 * Increment overrun to account for the lost events.
1893 			 */
1894 			local_add(page_entries, &cpu_buffer->overrun);
1895 			local_sub(rb_page_commit(to_remove_page), &cpu_buffer->entries_bytes);
1896 			local_inc(&cpu_buffer->pages_lost);
1897 		}
1898 
1899 		/*
1900 		 * We have already removed references to this list item, just
1901 		 * free up the buffer_page and its page
1902 		 */
1903 		free_buffer_page(to_remove_page);
1904 		nr_removed--;
1905 
1906 	} while (to_remove_page != last_page);
1907 
1908 	RB_WARN_ON(cpu_buffer, nr_removed);
1909 
1910 	return nr_removed == 0;
1911 }
1912 
1913 static bool
1914 rb_insert_pages(struct ring_buffer_per_cpu *cpu_buffer)
1915 {
1916 	struct list_head *pages = &cpu_buffer->new_pages;
1917 	unsigned long flags;
1918 	bool success;
1919 	int retries;
1920 
1921 	/* Can be called at early boot up, where interrupts must not been enabled */
1922 	raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
1923 	/*
1924 	 * We are holding the reader lock, so the reader page won't be swapped
1925 	 * in the ring buffer. Now we are racing with the writer trying to
1926 	 * move head page and the tail page.
1927 	 * We are going to adapt the reader page update process where:
1928 	 * 1. We first splice the start and end of list of new pages between
1929 	 *    the head page and its previous page.
1930 	 * 2. We cmpxchg the prev_page->next to point from head page to the
1931 	 *    start of new pages list.
1932 	 * 3. Finally, we update the head->prev to the end of new list.
1933 	 *
1934 	 * We will try this process 10 times, to make sure that we don't keep
1935 	 * spinning.
1936 	 */
1937 	retries = 10;
1938 	success = false;
1939 	while (retries--) {
1940 		struct list_head *head_page, *prev_page;
1941 		struct list_head *last_page, *first_page;
1942 		struct list_head *head_page_with_bit;
1943 		struct buffer_page *hpage = rb_set_head_page(cpu_buffer);
1944 
1945 		if (!hpage)
1946 			break;
1947 		head_page = &hpage->list;
1948 		prev_page = head_page->prev;
1949 
1950 		first_page = pages->next;
1951 		last_page  = pages->prev;
1952 
1953 		head_page_with_bit = (struct list_head *)
1954 				     ((unsigned long)head_page | RB_PAGE_HEAD);
1955 
1956 		last_page->next = head_page_with_bit;
1957 		first_page->prev = prev_page;
1958 
1959 		/* caution: head_page_with_bit gets updated on cmpxchg failure */
1960 		if (try_cmpxchg(&prev_page->next,
1961 				&head_page_with_bit, first_page)) {
1962 			/*
1963 			 * yay, we replaced the page pointer to our new list,
1964 			 * now, we just have to update to head page's prev
1965 			 * pointer to point to end of list
1966 			 */
1967 			head_page->prev = last_page;
1968 			success = true;
1969 			break;
1970 		}
1971 	}
1972 
1973 	if (success)
1974 		INIT_LIST_HEAD(pages);
1975 	/*
1976 	 * If we weren't successful in adding in new pages, warn and stop
1977 	 * tracing
1978 	 */
1979 	RB_WARN_ON(cpu_buffer, !success);
1980 	raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
1981 
1982 	/* free pages if they weren't inserted */
1983 	if (!success) {
1984 		struct buffer_page *bpage, *tmp;
1985 		list_for_each_entry_safe(bpage, tmp, &cpu_buffer->new_pages,
1986 					 list) {
1987 			list_del_init(&bpage->list);
1988 			free_buffer_page(bpage);
1989 		}
1990 	}
1991 	return success;
1992 }
1993 
1994 static void rb_update_pages(struct ring_buffer_per_cpu *cpu_buffer)
1995 {
1996 	bool success;
1997 
1998 	if (cpu_buffer->nr_pages_to_update > 0)
1999 		success = rb_insert_pages(cpu_buffer);
2000 	else
2001 		success = rb_remove_pages(cpu_buffer,
2002 					-cpu_buffer->nr_pages_to_update);
2003 
2004 	if (success)
2005 		cpu_buffer->nr_pages += cpu_buffer->nr_pages_to_update;
2006 }
2007 
2008 static void update_pages_handler(struct work_struct *work)
2009 {
2010 	struct ring_buffer_per_cpu *cpu_buffer = container_of(work,
2011 			struct ring_buffer_per_cpu, update_pages_work);
2012 	rb_update_pages(cpu_buffer);
2013 	complete(&cpu_buffer->update_done);
2014 }
2015 
2016 /**
2017  * ring_buffer_resize - resize the ring buffer
2018  * @buffer: the buffer to resize.
2019  * @size: the new size.
2020  * @cpu_id: the cpu buffer to resize
2021  *
2022  * Minimum size is 2 * buffer->subbuf_size.
2023  *
2024  * Returns 0 on success and < 0 on failure.
2025  */
2026 int ring_buffer_resize(struct trace_buffer *buffer, unsigned long size,
2027 			int cpu_id)
2028 {
2029 	struct ring_buffer_per_cpu *cpu_buffer;
2030 	unsigned long nr_pages;
2031 	int cpu, err;
2032 
2033 	/*
2034 	 * Always succeed at resizing a non-existent buffer:
2035 	 */
2036 	if (!buffer)
2037 		return 0;
2038 
2039 	/* Make sure the requested buffer exists */
2040 	if (cpu_id != RING_BUFFER_ALL_CPUS &&
2041 	    !cpumask_test_cpu(cpu_id, buffer->cpumask))
2042 		return 0;
2043 
2044 	nr_pages = DIV_ROUND_UP(size, buffer->subbuf_size);
2045 
2046 	/* we need a minimum of two pages */
2047 	if (nr_pages < 2)
2048 		nr_pages = 2;
2049 
2050 	/* prevent another thread from changing buffer sizes */
2051 	mutex_lock(&buffer->mutex);
2052 	atomic_inc(&buffer->resizing);
2053 
2054 	if (cpu_id == RING_BUFFER_ALL_CPUS) {
2055 		/*
2056 		 * Don't succeed if resizing is disabled, as a reader might be
2057 		 * manipulating the ring buffer and is expecting a sane state while
2058 		 * this is true.
2059 		 */
2060 		for_each_buffer_cpu(buffer, cpu) {
2061 			cpu_buffer = buffer->buffers[cpu];
2062 			if (atomic_read(&cpu_buffer->resize_disabled)) {
2063 				err = -EBUSY;
2064 				goto out_err_unlock;
2065 			}
2066 		}
2067 
2068 		/* calculate the pages to update */
2069 		for_each_buffer_cpu(buffer, cpu) {
2070 			cpu_buffer = buffer->buffers[cpu];
2071 
2072 			cpu_buffer->nr_pages_to_update = nr_pages -
2073 							cpu_buffer->nr_pages;
2074 			/*
2075 			 * nothing more to do for removing pages or no update
2076 			 */
2077 			if (cpu_buffer->nr_pages_to_update <= 0)
2078 				continue;
2079 			/*
2080 			 * to add pages, make sure all new pages can be
2081 			 * allocated without receiving ENOMEM
2082 			 */
2083 			INIT_LIST_HEAD(&cpu_buffer->new_pages);
2084 			if (__rb_allocate_pages(cpu_buffer, cpu_buffer->nr_pages_to_update,
2085 						&cpu_buffer->new_pages)) {
2086 				/* not enough memory for new pages */
2087 				err = -ENOMEM;
2088 				goto out_err;
2089 			}
2090 
2091 			cond_resched();
2092 		}
2093 
2094 		cpus_read_lock();
2095 		/*
2096 		 * Fire off all the required work handlers
2097 		 * We can't schedule on offline CPUs, but it's not necessary
2098 		 * since we can change their buffer sizes without any race.
2099 		 */
2100 		for_each_buffer_cpu(buffer, cpu) {
2101 			cpu_buffer = buffer->buffers[cpu];
2102 			if (!cpu_buffer->nr_pages_to_update)
2103 				continue;
2104 
2105 			/* Can't run something on an offline CPU. */
2106 			if (!cpu_online(cpu)) {
2107 				rb_update_pages(cpu_buffer);
2108 				cpu_buffer->nr_pages_to_update = 0;
2109 			} else {
2110 				/* Run directly if possible. */
2111 				migrate_disable();
2112 				if (cpu != smp_processor_id()) {
2113 					migrate_enable();
2114 					schedule_work_on(cpu,
2115 							 &cpu_buffer->update_pages_work);
2116 				} else {
2117 					update_pages_handler(&cpu_buffer->update_pages_work);
2118 					migrate_enable();
2119 				}
2120 			}
2121 		}
2122 
2123 		/* wait for all the updates to complete */
2124 		for_each_buffer_cpu(buffer, cpu) {
2125 			cpu_buffer = buffer->buffers[cpu];
2126 			if (!cpu_buffer->nr_pages_to_update)
2127 				continue;
2128 
2129 			if (cpu_online(cpu))
2130 				wait_for_completion(&cpu_buffer->update_done);
2131 			cpu_buffer->nr_pages_to_update = 0;
2132 		}
2133 
2134 		cpus_read_unlock();
2135 	} else {
2136 		cpu_buffer = buffer->buffers[cpu_id];
2137 
2138 		if (nr_pages == cpu_buffer->nr_pages)
2139 			goto out;
2140 
2141 		/*
2142 		 * Don't succeed if resizing is disabled, as a reader might be
2143 		 * manipulating the ring buffer and is expecting a sane state while
2144 		 * this is true.
2145 		 */
2146 		if (atomic_read(&cpu_buffer->resize_disabled)) {
2147 			err = -EBUSY;
2148 			goto out_err_unlock;
2149 		}
2150 
2151 		cpu_buffer->nr_pages_to_update = nr_pages -
2152 						cpu_buffer->nr_pages;
2153 
2154 		INIT_LIST_HEAD(&cpu_buffer->new_pages);
2155 		if (cpu_buffer->nr_pages_to_update > 0 &&
2156 			__rb_allocate_pages(cpu_buffer, cpu_buffer->nr_pages_to_update,
2157 					    &cpu_buffer->new_pages)) {
2158 			err = -ENOMEM;
2159 			goto out_err;
2160 		}
2161 
2162 		cpus_read_lock();
2163 
2164 		/* Can't run something on an offline CPU. */
2165 		if (!cpu_online(cpu_id))
2166 			rb_update_pages(cpu_buffer);
2167 		else {
2168 			/* Run directly if possible. */
2169 			migrate_disable();
2170 			if (cpu_id == smp_processor_id()) {
2171 				rb_update_pages(cpu_buffer);
2172 				migrate_enable();
2173 			} else {
2174 				migrate_enable();
2175 				schedule_work_on(cpu_id,
2176 						 &cpu_buffer->update_pages_work);
2177 				wait_for_completion(&cpu_buffer->update_done);
2178 			}
2179 		}
2180 
2181 		cpu_buffer->nr_pages_to_update = 0;
2182 		cpus_read_unlock();
2183 	}
2184 
2185  out:
2186 	/*
2187 	 * The ring buffer resize can happen with the ring buffer
2188 	 * enabled, so that the update disturbs the tracing as little
2189 	 * as possible. But if the buffer is disabled, we do not need
2190 	 * to worry about that, and we can take the time to verify
2191 	 * that the buffer is not corrupt.
2192 	 */
2193 	if (atomic_read(&buffer->record_disabled)) {
2194 		atomic_inc(&buffer->record_disabled);
2195 		/*
2196 		 * Even though the buffer was disabled, we must make sure
2197 		 * that it is truly disabled before calling rb_check_pages.
2198 		 * There could have been a race between checking
2199 		 * record_disable and incrementing it.
2200 		 */
2201 		synchronize_rcu();
2202 		for_each_buffer_cpu(buffer, cpu) {
2203 			cpu_buffer = buffer->buffers[cpu];
2204 			rb_check_pages(cpu_buffer);
2205 		}
2206 		atomic_dec(&buffer->record_disabled);
2207 	}
2208 
2209 	atomic_dec(&buffer->resizing);
2210 	mutex_unlock(&buffer->mutex);
2211 	return 0;
2212 
2213  out_err:
2214 	for_each_buffer_cpu(buffer, cpu) {
2215 		struct buffer_page *bpage, *tmp;
2216 
2217 		cpu_buffer = buffer->buffers[cpu];
2218 		cpu_buffer->nr_pages_to_update = 0;
2219 
2220 		if (list_empty(&cpu_buffer->new_pages))
2221 			continue;
2222 
2223 		list_for_each_entry_safe(bpage, tmp, &cpu_buffer->new_pages,
2224 					list) {
2225 			list_del_init(&bpage->list);
2226 			free_buffer_page(bpage);
2227 		}
2228 	}
2229  out_err_unlock:
2230 	atomic_dec(&buffer->resizing);
2231 	mutex_unlock(&buffer->mutex);
2232 	return err;
2233 }
2234 EXPORT_SYMBOL_GPL(ring_buffer_resize);
2235 
2236 void ring_buffer_change_overwrite(struct trace_buffer *buffer, int val)
2237 {
2238 	mutex_lock(&buffer->mutex);
2239 	if (val)
2240 		buffer->flags |= RB_FL_OVERWRITE;
2241 	else
2242 		buffer->flags &= ~RB_FL_OVERWRITE;
2243 	mutex_unlock(&buffer->mutex);
2244 }
2245 EXPORT_SYMBOL_GPL(ring_buffer_change_overwrite);
2246 
2247 static __always_inline void *__rb_page_index(struct buffer_page *bpage, unsigned index)
2248 {
2249 	return bpage->page->data + index;
2250 }
2251 
2252 static __always_inline struct ring_buffer_event *
2253 rb_reader_event(struct ring_buffer_per_cpu *cpu_buffer)
2254 {
2255 	return __rb_page_index(cpu_buffer->reader_page,
2256 			       cpu_buffer->reader_page->read);
2257 }
2258 
2259 static struct ring_buffer_event *
2260 rb_iter_head_event(struct ring_buffer_iter *iter)
2261 {
2262 	struct ring_buffer_event *event;
2263 	struct buffer_page *iter_head_page = iter->head_page;
2264 	unsigned long commit;
2265 	unsigned length;
2266 
2267 	if (iter->head != iter->next_event)
2268 		return iter->event;
2269 
2270 	/*
2271 	 * When the writer goes across pages, it issues a cmpxchg which
2272 	 * is a mb(), which will synchronize with the rmb here.
2273 	 * (see rb_tail_page_update() and __rb_reserve_next())
2274 	 */
2275 	commit = rb_page_commit(iter_head_page);
2276 	smp_rmb();
2277 
2278 	/* An event needs to be at least 8 bytes in size */
2279 	if (iter->head > commit - 8)
2280 		goto reset;
2281 
2282 	event = __rb_page_index(iter_head_page, iter->head);
2283 	length = rb_event_length(event);
2284 
2285 	/*
2286 	 * READ_ONCE() doesn't work on functions and we don't want the
2287 	 * compiler doing any crazy optimizations with length.
2288 	 */
2289 	barrier();
2290 
2291 	if ((iter->head + length) > commit || length > iter->event_size)
2292 		/* Writer corrupted the read? */
2293 		goto reset;
2294 
2295 	memcpy(iter->event, event, length);
2296 	/*
2297 	 * If the page stamp is still the same after this rmb() then the
2298 	 * event was safely copied without the writer entering the page.
2299 	 */
2300 	smp_rmb();
2301 
2302 	/* Make sure the page didn't change since we read this */
2303 	if (iter->page_stamp != iter_head_page->page->time_stamp ||
2304 	    commit > rb_page_commit(iter_head_page))
2305 		goto reset;
2306 
2307 	iter->next_event = iter->head + length;
2308 	return iter->event;
2309  reset:
2310 	/* Reset to the beginning */
2311 	iter->page_stamp = iter->read_stamp = iter->head_page->page->time_stamp;
2312 	iter->head = 0;
2313 	iter->next_event = 0;
2314 	iter->missed_events = 1;
2315 	return NULL;
2316 }
2317 
2318 /* Size is determined by what has been committed */
2319 static __always_inline unsigned rb_page_size(struct buffer_page *bpage)
2320 {
2321 	return rb_page_commit(bpage);
2322 }
2323 
2324 static __always_inline unsigned
2325 rb_commit_index(struct ring_buffer_per_cpu *cpu_buffer)
2326 {
2327 	return rb_page_commit(cpu_buffer->commit_page);
2328 }
2329 
2330 static __always_inline unsigned
2331 rb_event_index(struct ring_buffer_per_cpu *cpu_buffer, struct ring_buffer_event *event)
2332 {
2333 	unsigned long addr = (unsigned long)event;
2334 
2335 	addr &= (PAGE_SIZE << cpu_buffer->buffer->subbuf_order) - 1;
2336 
2337 	return addr - BUF_PAGE_HDR_SIZE;
2338 }
2339 
2340 static void rb_inc_iter(struct ring_buffer_iter *iter)
2341 {
2342 	struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
2343 
2344 	/*
2345 	 * The iterator could be on the reader page (it starts there).
2346 	 * But the head could have moved, since the reader was
2347 	 * found. Check for this case and assign the iterator
2348 	 * to the head page instead of next.
2349 	 */
2350 	if (iter->head_page == cpu_buffer->reader_page)
2351 		iter->head_page = rb_set_head_page(cpu_buffer);
2352 	else
2353 		rb_inc_page(&iter->head_page);
2354 
2355 	iter->page_stamp = iter->read_stamp = iter->head_page->page->time_stamp;
2356 	iter->head = 0;
2357 	iter->next_event = 0;
2358 }
2359 
2360 /*
2361  * rb_handle_head_page - writer hit the head page
2362  *
2363  * Returns: +1 to retry page
2364  *           0 to continue
2365  *          -1 on error
2366  */
2367 static int
2368 rb_handle_head_page(struct ring_buffer_per_cpu *cpu_buffer,
2369 		    struct buffer_page *tail_page,
2370 		    struct buffer_page *next_page)
2371 {
2372 	struct buffer_page *new_head;
2373 	int entries;
2374 	int type;
2375 	int ret;
2376 
2377 	entries = rb_page_entries(next_page);
2378 
2379 	/*
2380 	 * The hard part is here. We need to move the head
2381 	 * forward, and protect against both readers on
2382 	 * other CPUs and writers coming in via interrupts.
2383 	 */
2384 	type = rb_head_page_set_update(cpu_buffer, next_page, tail_page,
2385 				       RB_PAGE_HEAD);
2386 
2387 	/*
2388 	 * type can be one of four:
2389 	 *  NORMAL - an interrupt already moved it for us
2390 	 *  HEAD   - we are the first to get here.
2391 	 *  UPDATE - we are the interrupt interrupting
2392 	 *           a current move.
2393 	 *  MOVED  - a reader on another CPU moved the next
2394 	 *           pointer to its reader page. Give up
2395 	 *           and try again.
2396 	 */
2397 
2398 	switch (type) {
2399 	case RB_PAGE_HEAD:
2400 		/*
2401 		 * We changed the head to UPDATE, thus
2402 		 * it is our responsibility to update
2403 		 * the counters.
2404 		 */
2405 		local_add(entries, &cpu_buffer->overrun);
2406 		local_sub(rb_page_commit(next_page), &cpu_buffer->entries_bytes);
2407 		local_inc(&cpu_buffer->pages_lost);
2408 
2409 		/*
2410 		 * The entries will be zeroed out when we move the
2411 		 * tail page.
2412 		 */
2413 
2414 		/* still more to do */
2415 		break;
2416 
2417 	case RB_PAGE_UPDATE:
2418 		/*
2419 		 * This is an interrupt that interrupt the
2420 		 * previous update. Still more to do.
2421 		 */
2422 		break;
2423 	case RB_PAGE_NORMAL:
2424 		/*
2425 		 * An interrupt came in before the update
2426 		 * and processed this for us.
2427 		 * Nothing left to do.
2428 		 */
2429 		return 1;
2430 	case RB_PAGE_MOVED:
2431 		/*
2432 		 * The reader is on another CPU and just did
2433 		 * a swap with our next_page.
2434 		 * Try again.
2435 		 */
2436 		return 1;
2437 	default:
2438 		RB_WARN_ON(cpu_buffer, 1); /* WTF??? */
2439 		return -1;
2440 	}
2441 
2442 	/*
2443 	 * Now that we are here, the old head pointer is
2444 	 * set to UPDATE. This will keep the reader from
2445 	 * swapping the head page with the reader page.
2446 	 * The reader (on another CPU) will spin till
2447 	 * we are finished.
2448 	 *
2449 	 * We just need to protect against interrupts
2450 	 * doing the job. We will set the next pointer
2451 	 * to HEAD. After that, we set the old pointer
2452 	 * to NORMAL, but only if it was HEAD before.
2453 	 * otherwise we are an interrupt, and only
2454 	 * want the outer most commit to reset it.
2455 	 */
2456 	new_head = next_page;
2457 	rb_inc_page(&new_head);
2458 
2459 	ret = rb_head_page_set_head(cpu_buffer, new_head, next_page,
2460 				    RB_PAGE_NORMAL);
2461 
2462 	/*
2463 	 * Valid returns are:
2464 	 *  HEAD   - an interrupt came in and already set it.
2465 	 *  NORMAL - One of two things:
2466 	 *            1) We really set it.
2467 	 *            2) A bunch of interrupts came in and moved
2468 	 *               the page forward again.
2469 	 */
2470 	switch (ret) {
2471 	case RB_PAGE_HEAD:
2472 	case RB_PAGE_NORMAL:
2473 		/* OK */
2474 		break;
2475 	default:
2476 		RB_WARN_ON(cpu_buffer, 1);
2477 		return -1;
2478 	}
2479 
2480 	/*
2481 	 * It is possible that an interrupt came in,
2482 	 * set the head up, then more interrupts came in
2483 	 * and moved it again. When we get back here,
2484 	 * the page would have been set to NORMAL but we
2485 	 * just set it back to HEAD.
2486 	 *
2487 	 * How do you detect this? Well, if that happened
2488 	 * the tail page would have moved.
2489 	 */
2490 	if (ret == RB_PAGE_NORMAL) {
2491 		struct buffer_page *buffer_tail_page;
2492 
2493 		buffer_tail_page = READ_ONCE(cpu_buffer->tail_page);
2494 		/*
2495 		 * If the tail had moved passed next, then we need
2496 		 * to reset the pointer.
2497 		 */
2498 		if (buffer_tail_page != tail_page &&
2499 		    buffer_tail_page != next_page)
2500 			rb_head_page_set_normal(cpu_buffer, new_head,
2501 						next_page,
2502 						RB_PAGE_HEAD);
2503 	}
2504 
2505 	/*
2506 	 * If this was the outer most commit (the one that
2507 	 * changed the original pointer from HEAD to UPDATE),
2508 	 * then it is up to us to reset it to NORMAL.
2509 	 */
2510 	if (type == RB_PAGE_HEAD) {
2511 		ret = rb_head_page_set_normal(cpu_buffer, next_page,
2512 					      tail_page,
2513 					      RB_PAGE_UPDATE);
2514 		if (RB_WARN_ON(cpu_buffer,
2515 			       ret != RB_PAGE_UPDATE))
2516 			return -1;
2517 	}
2518 
2519 	return 0;
2520 }
2521 
2522 static inline void
2523 rb_reset_tail(struct ring_buffer_per_cpu *cpu_buffer,
2524 	      unsigned long tail, struct rb_event_info *info)
2525 {
2526 	unsigned long bsize = READ_ONCE(cpu_buffer->buffer->subbuf_size);
2527 	struct buffer_page *tail_page = info->tail_page;
2528 	struct ring_buffer_event *event;
2529 	unsigned long length = info->length;
2530 
2531 	/*
2532 	 * Only the event that crossed the page boundary
2533 	 * must fill the old tail_page with padding.
2534 	 */
2535 	if (tail >= bsize) {
2536 		/*
2537 		 * If the page was filled, then we still need
2538 		 * to update the real_end. Reset it to zero
2539 		 * and the reader will ignore it.
2540 		 */
2541 		if (tail == bsize)
2542 			tail_page->real_end = 0;
2543 
2544 		local_sub(length, &tail_page->write);
2545 		return;
2546 	}
2547 
2548 	event = __rb_page_index(tail_page, tail);
2549 
2550 	/*
2551 	 * Save the original length to the meta data.
2552 	 * This will be used by the reader to add lost event
2553 	 * counter.
2554 	 */
2555 	tail_page->real_end = tail;
2556 
2557 	/*
2558 	 * If this event is bigger than the minimum size, then
2559 	 * we need to be careful that we don't subtract the
2560 	 * write counter enough to allow another writer to slip
2561 	 * in on this page.
2562 	 * We put in a discarded commit instead, to make sure
2563 	 * that this space is not used again, and this space will
2564 	 * not be accounted into 'entries_bytes'.
2565 	 *
2566 	 * If we are less than the minimum size, we don't need to
2567 	 * worry about it.
2568 	 */
2569 	if (tail > (bsize - RB_EVNT_MIN_SIZE)) {
2570 		/* No room for any events */
2571 
2572 		/* Mark the rest of the page with padding */
2573 		rb_event_set_padding(event);
2574 
2575 		/* Make sure the padding is visible before the write update */
2576 		smp_wmb();
2577 
2578 		/* Set the write back to the previous setting */
2579 		local_sub(length, &tail_page->write);
2580 		return;
2581 	}
2582 
2583 	/* Put in a discarded event */
2584 	event->array[0] = (bsize - tail) - RB_EVNT_HDR_SIZE;
2585 	event->type_len = RINGBUF_TYPE_PADDING;
2586 	/* time delta must be non zero */
2587 	event->time_delta = 1;
2588 
2589 	/* account for padding bytes */
2590 	local_add(bsize - tail, &cpu_buffer->entries_bytes);
2591 
2592 	/* Make sure the padding is visible before the tail_page->write update */
2593 	smp_wmb();
2594 
2595 	/* Set write to end of buffer */
2596 	length = (tail + length) - bsize;
2597 	local_sub(length, &tail_page->write);
2598 }
2599 
2600 static inline void rb_end_commit(struct ring_buffer_per_cpu *cpu_buffer);
2601 
2602 /*
2603  * This is the slow path, force gcc not to inline it.
2604  */
2605 static noinline struct ring_buffer_event *
2606 rb_move_tail(struct ring_buffer_per_cpu *cpu_buffer,
2607 	     unsigned long tail, struct rb_event_info *info)
2608 {
2609 	struct buffer_page *tail_page = info->tail_page;
2610 	struct buffer_page *commit_page = cpu_buffer->commit_page;
2611 	struct trace_buffer *buffer = cpu_buffer->buffer;
2612 	struct buffer_page *next_page;
2613 	int ret;
2614 
2615 	next_page = tail_page;
2616 
2617 	rb_inc_page(&next_page);
2618 
2619 	/*
2620 	 * If for some reason, we had an interrupt storm that made
2621 	 * it all the way around the buffer, bail, and warn
2622 	 * about it.
2623 	 */
2624 	if (unlikely(next_page == commit_page)) {
2625 		local_inc(&cpu_buffer->commit_overrun);
2626 		goto out_reset;
2627 	}
2628 
2629 	/*
2630 	 * This is where the fun begins!
2631 	 *
2632 	 * We are fighting against races between a reader that
2633 	 * could be on another CPU trying to swap its reader
2634 	 * page with the buffer head.
2635 	 *
2636 	 * We are also fighting against interrupts coming in and
2637 	 * moving the head or tail on us as well.
2638 	 *
2639 	 * If the next page is the head page then we have filled
2640 	 * the buffer, unless the commit page is still on the
2641 	 * reader page.
2642 	 */
2643 	if (rb_is_head_page(next_page, &tail_page->list)) {
2644 
2645 		/*
2646 		 * If the commit is not on the reader page, then
2647 		 * move the header page.
2648 		 */
2649 		if (!rb_is_reader_page(cpu_buffer->commit_page)) {
2650 			/*
2651 			 * If we are not in overwrite mode,
2652 			 * this is easy, just stop here.
2653 			 */
2654 			if (!(buffer->flags & RB_FL_OVERWRITE)) {
2655 				local_inc(&cpu_buffer->dropped_events);
2656 				goto out_reset;
2657 			}
2658 
2659 			ret = rb_handle_head_page(cpu_buffer,
2660 						  tail_page,
2661 						  next_page);
2662 			if (ret < 0)
2663 				goto out_reset;
2664 			if (ret)
2665 				goto out_again;
2666 		} else {
2667 			/*
2668 			 * We need to be careful here too. The
2669 			 * commit page could still be on the reader
2670 			 * page. We could have a small buffer, and
2671 			 * have filled up the buffer with events
2672 			 * from interrupts and such, and wrapped.
2673 			 *
2674 			 * Note, if the tail page is also on the
2675 			 * reader_page, we let it move out.
2676 			 */
2677 			if (unlikely((cpu_buffer->commit_page !=
2678 				      cpu_buffer->tail_page) &&
2679 				     (cpu_buffer->commit_page ==
2680 				      cpu_buffer->reader_page))) {
2681 				local_inc(&cpu_buffer->commit_overrun);
2682 				goto out_reset;
2683 			}
2684 		}
2685 	}
2686 
2687 	rb_tail_page_update(cpu_buffer, tail_page, next_page);
2688 
2689  out_again:
2690 
2691 	rb_reset_tail(cpu_buffer, tail, info);
2692 
2693 	/* Commit what we have for now. */
2694 	rb_end_commit(cpu_buffer);
2695 	/* rb_end_commit() decs committing */
2696 	local_inc(&cpu_buffer->committing);
2697 
2698 	/* fail and let the caller try again */
2699 	return ERR_PTR(-EAGAIN);
2700 
2701  out_reset:
2702 	/* reset write */
2703 	rb_reset_tail(cpu_buffer, tail, info);
2704 
2705 	return NULL;
2706 }
2707 
2708 /* Slow path */
2709 static struct ring_buffer_event *
2710 rb_add_time_stamp(struct ring_buffer_per_cpu *cpu_buffer,
2711 		  struct ring_buffer_event *event, u64 delta, bool abs)
2712 {
2713 	if (abs)
2714 		event->type_len = RINGBUF_TYPE_TIME_STAMP;
2715 	else
2716 		event->type_len = RINGBUF_TYPE_TIME_EXTEND;
2717 
2718 	/* Not the first event on the page, or not delta? */
2719 	if (abs || rb_event_index(cpu_buffer, event)) {
2720 		event->time_delta = delta & TS_MASK;
2721 		event->array[0] = delta >> TS_SHIFT;
2722 	} else {
2723 		/* nope, just zero it */
2724 		event->time_delta = 0;
2725 		event->array[0] = 0;
2726 	}
2727 
2728 	return skip_time_extend(event);
2729 }
2730 
2731 #ifndef CONFIG_HAVE_UNSTABLE_SCHED_CLOCK
2732 static inline bool sched_clock_stable(void)
2733 {
2734 	return true;
2735 }
2736 #endif
2737 
2738 static void
2739 rb_check_timestamp(struct ring_buffer_per_cpu *cpu_buffer,
2740 		   struct rb_event_info *info)
2741 {
2742 	u64 write_stamp;
2743 
2744 	WARN_ONCE(1, "Delta way too big! %llu ts=%llu before=%llu after=%llu write stamp=%llu\n%s",
2745 		  (unsigned long long)info->delta,
2746 		  (unsigned long long)info->ts,
2747 		  (unsigned long long)info->before,
2748 		  (unsigned long long)info->after,
2749 		  (unsigned long long)({rb_time_read(&cpu_buffer->write_stamp, &write_stamp); write_stamp;}),
2750 		  sched_clock_stable() ? "" :
2751 		  "If you just came from a suspend/resume,\n"
2752 		  "please switch to the trace global clock:\n"
2753 		  "  echo global > /sys/kernel/tracing/trace_clock\n"
2754 		  "or add trace_clock=global to the kernel command line\n");
2755 }
2756 
2757 static void rb_add_timestamp(struct ring_buffer_per_cpu *cpu_buffer,
2758 				      struct ring_buffer_event **event,
2759 				      struct rb_event_info *info,
2760 				      u64 *delta,
2761 				      unsigned int *length)
2762 {
2763 	bool abs = info->add_timestamp &
2764 		(RB_ADD_STAMP_FORCE | RB_ADD_STAMP_ABSOLUTE);
2765 
2766 	if (unlikely(info->delta > (1ULL << 59))) {
2767 		/*
2768 		 * Some timers can use more than 59 bits, and when a timestamp
2769 		 * is added to the buffer, it will lose those bits.
2770 		 */
2771 		if (abs && (info->ts & TS_MSB)) {
2772 			info->delta &= ABS_TS_MASK;
2773 
2774 		/* did the clock go backwards */
2775 		} else if (info->before == info->after && info->before > info->ts) {
2776 			/* not interrupted */
2777 			static int once;
2778 
2779 			/*
2780 			 * This is possible with a recalibrating of the TSC.
2781 			 * Do not produce a call stack, but just report it.
2782 			 */
2783 			if (!once) {
2784 				once++;
2785 				pr_warn("Ring buffer clock went backwards: %llu -> %llu\n",
2786 					info->before, info->ts);
2787 			}
2788 		} else
2789 			rb_check_timestamp(cpu_buffer, info);
2790 		if (!abs)
2791 			info->delta = 0;
2792 	}
2793 	*event = rb_add_time_stamp(cpu_buffer, *event, info->delta, abs);
2794 	*length -= RB_LEN_TIME_EXTEND;
2795 	*delta = 0;
2796 }
2797 
2798 /**
2799  * rb_update_event - update event type and data
2800  * @cpu_buffer: The per cpu buffer of the @event
2801  * @event: the event to update
2802  * @info: The info to update the @event with (contains length and delta)
2803  *
2804  * Update the type and data fields of the @event. The length
2805  * is the actual size that is written to the ring buffer,
2806  * and with this, we can determine what to place into the
2807  * data field.
2808  */
2809 static void
2810 rb_update_event(struct ring_buffer_per_cpu *cpu_buffer,
2811 		struct ring_buffer_event *event,
2812 		struct rb_event_info *info)
2813 {
2814 	unsigned length = info->length;
2815 	u64 delta = info->delta;
2816 	unsigned int nest = local_read(&cpu_buffer->committing) - 1;
2817 
2818 	if (!WARN_ON_ONCE(nest >= MAX_NEST))
2819 		cpu_buffer->event_stamp[nest] = info->ts;
2820 
2821 	/*
2822 	 * If we need to add a timestamp, then we
2823 	 * add it to the start of the reserved space.
2824 	 */
2825 	if (unlikely(info->add_timestamp))
2826 		rb_add_timestamp(cpu_buffer, &event, info, &delta, &length);
2827 
2828 	event->time_delta = delta;
2829 	length -= RB_EVNT_HDR_SIZE;
2830 	if (length > RB_MAX_SMALL_DATA || RB_FORCE_8BYTE_ALIGNMENT) {
2831 		event->type_len = 0;
2832 		event->array[0] = length;
2833 	} else
2834 		event->type_len = DIV_ROUND_UP(length, RB_ALIGNMENT);
2835 }
2836 
2837 static unsigned rb_calculate_event_length(unsigned length)
2838 {
2839 	struct ring_buffer_event event; /* Used only for sizeof array */
2840 
2841 	/* zero length can cause confusions */
2842 	if (!length)
2843 		length++;
2844 
2845 	if (length > RB_MAX_SMALL_DATA || RB_FORCE_8BYTE_ALIGNMENT)
2846 		length += sizeof(event.array[0]);
2847 
2848 	length += RB_EVNT_HDR_SIZE;
2849 	length = ALIGN(length, RB_ARCH_ALIGNMENT);
2850 
2851 	/*
2852 	 * In case the time delta is larger than the 27 bits for it
2853 	 * in the header, we need to add a timestamp. If another
2854 	 * event comes in when trying to discard this one to increase
2855 	 * the length, then the timestamp will be added in the allocated
2856 	 * space of this event. If length is bigger than the size needed
2857 	 * for the TIME_EXTEND, then padding has to be used. The events
2858 	 * length must be either RB_LEN_TIME_EXTEND, or greater than or equal
2859 	 * to RB_LEN_TIME_EXTEND + 8, as 8 is the minimum size for padding.
2860 	 * As length is a multiple of 4, we only need to worry if it
2861 	 * is 12 (RB_LEN_TIME_EXTEND + 4).
2862 	 */
2863 	if (length == RB_LEN_TIME_EXTEND + RB_ALIGNMENT)
2864 		length += RB_ALIGNMENT;
2865 
2866 	return length;
2867 }
2868 
2869 static inline bool
2870 rb_try_to_discard(struct ring_buffer_per_cpu *cpu_buffer,
2871 		  struct ring_buffer_event *event)
2872 {
2873 	unsigned long new_index, old_index;
2874 	struct buffer_page *bpage;
2875 	unsigned long addr;
2876 
2877 	new_index = rb_event_index(cpu_buffer, event);
2878 	old_index = new_index + rb_event_ts_length(event);
2879 	addr = (unsigned long)event;
2880 	addr &= ~((PAGE_SIZE << cpu_buffer->buffer->subbuf_order) - 1);
2881 
2882 	bpage = READ_ONCE(cpu_buffer->tail_page);
2883 
2884 	/*
2885 	 * Make sure the tail_page is still the same and
2886 	 * the next write location is the end of this event
2887 	 */
2888 	if (bpage->page == (void *)addr && rb_page_write(bpage) == old_index) {
2889 		unsigned long write_mask =
2890 			local_read(&bpage->write) & ~RB_WRITE_MASK;
2891 		unsigned long event_length = rb_event_length(event);
2892 
2893 		/*
2894 		 * For the before_stamp to be different than the write_stamp
2895 		 * to make sure that the next event adds an absolute
2896 		 * value and does not rely on the saved write stamp, which
2897 		 * is now going to be bogus.
2898 		 *
2899 		 * By setting the before_stamp to zero, the next event
2900 		 * is not going to use the write_stamp and will instead
2901 		 * create an absolute timestamp. This means there's no
2902 		 * reason to update the wirte_stamp!
2903 		 */
2904 		rb_time_set(&cpu_buffer->before_stamp, 0);
2905 
2906 		/*
2907 		 * If an event were to come in now, it would see that the
2908 		 * write_stamp and the before_stamp are different, and assume
2909 		 * that this event just added itself before updating
2910 		 * the write stamp. The interrupting event will fix the
2911 		 * write stamp for us, and use an absolute timestamp.
2912 		 */
2913 
2914 		/*
2915 		 * This is on the tail page. It is possible that
2916 		 * a write could come in and move the tail page
2917 		 * and write to the next page. That is fine
2918 		 * because we just shorten what is on this page.
2919 		 */
2920 		old_index += write_mask;
2921 		new_index += write_mask;
2922 
2923 		/* caution: old_index gets updated on cmpxchg failure */
2924 		if (local_try_cmpxchg(&bpage->write, &old_index, new_index)) {
2925 			/* update counters */
2926 			local_sub(event_length, &cpu_buffer->entries_bytes);
2927 			return true;
2928 		}
2929 	}
2930 
2931 	/* could not discard */
2932 	return false;
2933 }
2934 
2935 static void rb_start_commit(struct ring_buffer_per_cpu *cpu_buffer)
2936 {
2937 	local_inc(&cpu_buffer->committing);
2938 	local_inc(&cpu_buffer->commits);
2939 }
2940 
2941 static __always_inline void
2942 rb_set_commit_to_write(struct ring_buffer_per_cpu *cpu_buffer)
2943 {
2944 	unsigned long max_count;
2945 
2946 	/*
2947 	 * We only race with interrupts and NMIs on this CPU.
2948 	 * If we own the commit event, then we can commit
2949 	 * all others that interrupted us, since the interruptions
2950 	 * are in stack format (they finish before they come
2951 	 * back to us). This allows us to do a simple loop to
2952 	 * assign the commit to the tail.
2953 	 */
2954  again:
2955 	max_count = cpu_buffer->nr_pages * 100;
2956 
2957 	while (cpu_buffer->commit_page != READ_ONCE(cpu_buffer->tail_page)) {
2958 		if (RB_WARN_ON(cpu_buffer, !(--max_count)))
2959 			return;
2960 		if (RB_WARN_ON(cpu_buffer,
2961 			       rb_is_reader_page(cpu_buffer->tail_page)))
2962 			return;
2963 		/*
2964 		 * No need for a memory barrier here, as the update
2965 		 * of the tail_page did it for this page.
2966 		 */
2967 		local_set(&cpu_buffer->commit_page->page->commit,
2968 			  rb_page_write(cpu_buffer->commit_page));
2969 		rb_inc_page(&cpu_buffer->commit_page);
2970 		/* add barrier to keep gcc from optimizing too much */
2971 		barrier();
2972 	}
2973 	while (rb_commit_index(cpu_buffer) !=
2974 	       rb_page_write(cpu_buffer->commit_page)) {
2975 
2976 		/* Make sure the readers see the content of what is committed. */
2977 		smp_wmb();
2978 		local_set(&cpu_buffer->commit_page->page->commit,
2979 			  rb_page_write(cpu_buffer->commit_page));
2980 		RB_WARN_ON(cpu_buffer,
2981 			   local_read(&cpu_buffer->commit_page->page->commit) &
2982 			   ~RB_WRITE_MASK);
2983 		barrier();
2984 	}
2985 
2986 	/* again, keep gcc from optimizing */
2987 	barrier();
2988 
2989 	/*
2990 	 * If an interrupt came in just after the first while loop
2991 	 * and pushed the tail page forward, we will be left with
2992 	 * a dangling commit that will never go forward.
2993 	 */
2994 	if (unlikely(cpu_buffer->commit_page != READ_ONCE(cpu_buffer->tail_page)))
2995 		goto again;
2996 }
2997 
2998 static __always_inline void rb_end_commit(struct ring_buffer_per_cpu *cpu_buffer)
2999 {
3000 	unsigned long commits;
3001 
3002 	if (RB_WARN_ON(cpu_buffer,
3003 		       !local_read(&cpu_buffer->committing)))
3004 		return;
3005 
3006  again:
3007 	commits = local_read(&cpu_buffer->commits);
3008 	/* synchronize with interrupts */
3009 	barrier();
3010 	if (local_read(&cpu_buffer->committing) == 1)
3011 		rb_set_commit_to_write(cpu_buffer);
3012 
3013 	local_dec(&cpu_buffer->committing);
3014 
3015 	/* synchronize with interrupts */
3016 	barrier();
3017 
3018 	/*
3019 	 * Need to account for interrupts coming in between the
3020 	 * updating of the commit page and the clearing of the
3021 	 * committing counter.
3022 	 */
3023 	if (unlikely(local_read(&cpu_buffer->commits) != commits) &&
3024 	    !local_read(&cpu_buffer->committing)) {
3025 		local_inc(&cpu_buffer->committing);
3026 		goto again;
3027 	}
3028 }
3029 
3030 static inline void rb_event_discard(struct ring_buffer_event *event)
3031 {
3032 	if (extended_time(event))
3033 		event = skip_time_extend(event);
3034 
3035 	/* array[0] holds the actual length for the discarded event */
3036 	event->array[0] = rb_event_data_length(event) - RB_EVNT_HDR_SIZE;
3037 	event->type_len = RINGBUF_TYPE_PADDING;
3038 	/* time delta must be non zero */
3039 	if (!event->time_delta)
3040 		event->time_delta = 1;
3041 }
3042 
3043 static void rb_commit(struct ring_buffer_per_cpu *cpu_buffer)
3044 {
3045 	local_inc(&cpu_buffer->entries);
3046 	rb_end_commit(cpu_buffer);
3047 }
3048 
3049 static __always_inline void
3050 rb_wakeups(struct trace_buffer *buffer, struct ring_buffer_per_cpu *cpu_buffer)
3051 {
3052 	if (buffer->irq_work.waiters_pending) {
3053 		buffer->irq_work.waiters_pending = false;
3054 		/* irq_work_queue() supplies it's own memory barriers */
3055 		irq_work_queue(&buffer->irq_work.work);
3056 	}
3057 
3058 	if (cpu_buffer->irq_work.waiters_pending) {
3059 		cpu_buffer->irq_work.waiters_pending = false;
3060 		/* irq_work_queue() supplies it's own memory barriers */
3061 		irq_work_queue(&cpu_buffer->irq_work.work);
3062 	}
3063 
3064 	if (cpu_buffer->last_pages_touch == local_read(&cpu_buffer->pages_touched))
3065 		return;
3066 
3067 	if (cpu_buffer->reader_page == cpu_buffer->commit_page)
3068 		return;
3069 
3070 	if (!cpu_buffer->irq_work.full_waiters_pending)
3071 		return;
3072 
3073 	cpu_buffer->last_pages_touch = local_read(&cpu_buffer->pages_touched);
3074 
3075 	if (!full_hit(buffer, cpu_buffer->cpu, cpu_buffer->shortest_full))
3076 		return;
3077 
3078 	cpu_buffer->irq_work.wakeup_full = true;
3079 	cpu_buffer->irq_work.full_waiters_pending = false;
3080 	/* irq_work_queue() supplies it's own memory barriers */
3081 	irq_work_queue(&cpu_buffer->irq_work.work);
3082 }
3083 
3084 #ifdef CONFIG_RING_BUFFER_RECORD_RECURSION
3085 # define do_ring_buffer_record_recursion()	\
3086 	do_ftrace_record_recursion(_THIS_IP_, _RET_IP_)
3087 #else
3088 # define do_ring_buffer_record_recursion() do { } while (0)
3089 #endif
3090 
3091 /*
3092  * The lock and unlock are done within a preempt disable section.
3093  * The current_context per_cpu variable can only be modified
3094  * by the current task between lock and unlock. But it can
3095  * be modified more than once via an interrupt. To pass this
3096  * information from the lock to the unlock without having to
3097  * access the 'in_interrupt()' functions again (which do show
3098  * a bit of overhead in something as critical as function tracing,
3099  * we use a bitmask trick.
3100  *
3101  *  bit 1 =  NMI context
3102  *  bit 2 =  IRQ context
3103  *  bit 3 =  SoftIRQ context
3104  *  bit 4 =  normal context.
3105  *
3106  * This works because this is the order of contexts that can
3107  * preempt other contexts. A SoftIRQ never preempts an IRQ
3108  * context.
3109  *
3110  * When the context is determined, the corresponding bit is
3111  * checked and set (if it was set, then a recursion of that context
3112  * happened).
3113  *
3114  * On unlock, we need to clear this bit. To do so, just subtract
3115  * 1 from the current_context and AND it to itself.
3116  *
3117  * (binary)
3118  *  101 - 1 = 100
3119  *  101 & 100 = 100 (clearing bit zero)
3120  *
3121  *  1010 - 1 = 1001
3122  *  1010 & 1001 = 1000 (clearing bit 1)
3123  *
3124  * The least significant bit can be cleared this way, and it
3125  * just so happens that it is the same bit corresponding to
3126  * the current context.
3127  *
3128  * Now the TRANSITION bit breaks the above slightly. The TRANSITION bit
3129  * is set when a recursion is detected at the current context, and if
3130  * the TRANSITION bit is already set, it will fail the recursion.
3131  * This is needed because there's a lag between the changing of
3132  * interrupt context and updating the preempt count. In this case,
3133  * a false positive will be found. To handle this, one extra recursion
3134  * is allowed, and this is done by the TRANSITION bit. If the TRANSITION
3135  * bit is already set, then it is considered a recursion and the function
3136  * ends. Otherwise, the TRANSITION bit is set, and that bit is returned.
3137  *
3138  * On the trace_recursive_unlock(), the TRANSITION bit will be the first
3139  * to be cleared. Even if it wasn't the context that set it. That is,
3140  * if an interrupt comes in while NORMAL bit is set and the ring buffer
3141  * is called before preempt_count() is updated, since the check will
3142  * be on the NORMAL bit, the TRANSITION bit will then be set. If an
3143  * NMI then comes in, it will set the NMI bit, but when the NMI code
3144  * does the trace_recursive_unlock() it will clear the TRANSITION bit
3145  * and leave the NMI bit set. But this is fine, because the interrupt
3146  * code that set the TRANSITION bit will then clear the NMI bit when it
3147  * calls trace_recursive_unlock(). If another NMI comes in, it will
3148  * set the TRANSITION bit and continue.
3149  *
3150  * Note: The TRANSITION bit only handles a single transition between context.
3151  */
3152 
3153 static __always_inline bool
3154 trace_recursive_lock(struct ring_buffer_per_cpu *cpu_buffer)
3155 {
3156 	unsigned int val = cpu_buffer->current_context;
3157 	int bit = interrupt_context_level();
3158 
3159 	bit = RB_CTX_NORMAL - bit;
3160 
3161 	if (unlikely(val & (1 << (bit + cpu_buffer->nest)))) {
3162 		/*
3163 		 * It is possible that this was called by transitioning
3164 		 * between interrupt context, and preempt_count() has not
3165 		 * been updated yet. In this case, use the TRANSITION bit.
3166 		 */
3167 		bit = RB_CTX_TRANSITION;
3168 		if (val & (1 << (bit + cpu_buffer->nest))) {
3169 			do_ring_buffer_record_recursion();
3170 			return true;
3171 		}
3172 	}
3173 
3174 	val |= (1 << (bit + cpu_buffer->nest));
3175 	cpu_buffer->current_context = val;
3176 
3177 	return false;
3178 }
3179 
3180 static __always_inline void
3181 trace_recursive_unlock(struct ring_buffer_per_cpu *cpu_buffer)
3182 {
3183 	cpu_buffer->current_context &=
3184 		cpu_buffer->current_context - (1 << cpu_buffer->nest);
3185 }
3186 
3187 /* The recursive locking above uses 5 bits */
3188 #define NESTED_BITS 5
3189 
3190 /**
3191  * ring_buffer_nest_start - Allow to trace while nested
3192  * @buffer: The ring buffer to modify
3193  *
3194  * The ring buffer has a safety mechanism to prevent recursion.
3195  * But there may be a case where a trace needs to be done while
3196  * tracing something else. In this case, calling this function
3197  * will allow this function to nest within a currently active
3198  * ring_buffer_lock_reserve().
3199  *
3200  * Call this function before calling another ring_buffer_lock_reserve() and
3201  * call ring_buffer_nest_end() after the nested ring_buffer_unlock_commit().
3202  */
3203 void ring_buffer_nest_start(struct trace_buffer *buffer)
3204 {
3205 	struct ring_buffer_per_cpu *cpu_buffer;
3206 	int cpu;
3207 
3208 	/* Enabled by ring_buffer_nest_end() */
3209 	preempt_disable_notrace();
3210 	cpu = raw_smp_processor_id();
3211 	cpu_buffer = buffer->buffers[cpu];
3212 	/* This is the shift value for the above recursive locking */
3213 	cpu_buffer->nest += NESTED_BITS;
3214 }
3215 
3216 /**
3217  * ring_buffer_nest_end - Allow to trace while nested
3218  * @buffer: The ring buffer to modify
3219  *
3220  * Must be called after ring_buffer_nest_start() and after the
3221  * ring_buffer_unlock_commit().
3222  */
3223 void ring_buffer_nest_end(struct trace_buffer *buffer)
3224 {
3225 	struct ring_buffer_per_cpu *cpu_buffer;
3226 	int cpu;
3227 
3228 	/* disabled by ring_buffer_nest_start() */
3229 	cpu = raw_smp_processor_id();
3230 	cpu_buffer = buffer->buffers[cpu];
3231 	/* This is the shift value for the above recursive locking */
3232 	cpu_buffer->nest -= NESTED_BITS;
3233 	preempt_enable_notrace();
3234 }
3235 
3236 /**
3237  * ring_buffer_unlock_commit - commit a reserved
3238  * @buffer: The buffer to commit to
3239  *
3240  * This commits the data to the ring buffer, and releases any locks held.
3241  *
3242  * Must be paired with ring_buffer_lock_reserve.
3243  */
3244 int ring_buffer_unlock_commit(struct trace_buffer *buffer)
3245 {
3246 	struct ring_buffer_per_cpu *cpu_buffer;
3247 	int cpu = raw_smp_processor_id();
3248 
3249 	cpu_buffer = buffer->buffers[cpu];
3250 
3251 	rb_commit(cpu_buffer);
3252 
3253 	rb_wakeups(buffer, cpu_buffer);
3254 
3255 	trace_recursive_unlock(cpu_buffer);
3256 
3257 	preempt_enable_notrace();
3258 
3259 	return 0;
3260 }
3261 EXPORT_SYMBOL_GPL(ring_buffer_unlock_commit);
3262 
3263 /* Special value to validate all deltas on a page. */
3264 #define CHECK_FULL_PAGE		1L
3265 
3266 #ifdef CONFIG_RING_BUFFER_VALIDATE_TIME_DELTAS
3267 
3268 static const char *show_irq_str(int bits)
3269 {
3270 	const char *type[] = {
3271 		".",	// 0
3272 		"s",	// 1
3273 		"h",	// 2
3274 		"Hs",	// 3
3275 		"n",	// 4
3276 		"Ns",	// 5
3277 		"Nh",	// 6
3278 		"NHs",	// 7
3279 	};
3280 
3281 	return type[bits];
3282 }
3283 
3284 /* Assume this is an trace event */
3285 static const char *show_flags(struct ring_buffer_event *event)
3286 {
3287 	struct trace_entry *entry;
3288 	int bits = 0;
3289 
3290 	if (rb_event_data_length(event) - RB_EVNT_HDR_SIZE < sizeof(*entry))
3291 		return "X";
3292 
3293 	entry = ring_buffer_event_data(event);
3294 
3295 	if (entry->flags & TRACE_FLAG_SOFTIRQ)
3296 		bits |= 1;
3297 
3298 	if (entry->flags & TRACE_FLAG_HARDIRQ)
3299 		bits |= 2;
3300 
3301 	if (entry->flags & TRACE_FLAG_NMI)
3302 		bits |= 4;
3303 
3304 	return show_irq_str(bits);
3305 }
3306 
3307 static const char *show_irq(struct ring_buffer_event *event)
3308 {
3309 	struct trace_entry *entry;
3310 
3311 	if (rb_event_data_length(event) - RB_EVNT_HDR_SIZE < sizeof(*entry))
3312 		return "";
3313 
3314 	entry = ring_buffer_event_data(event);
3315 	if (entry->flags & TRACE_FLAG_IRQS_OFF)
3316 		return "d";
3317 	return "";
3318 }
3319 
3320 static const char *show_interrupt_level(void)
3321 {
3322 	unsigned long pc = preempt_count();
3323 	unsigned char level = 0;
3324 
3325 	if (pc & SOFTIRQ_OFFSET)
3326 		level |= 1;
3327 
3328 	if (pc & HARDIRQ_MASK)
3329 		level |= 2;
3330 
3331 	if (pc & NMI_MASK)
3332 		level |= 4;
3333 
3334 	return show_irq_str(level);
3335 }
3336 
3337 static void dump_buffer_page(struct buffer_data_page *bpage,
3338 			     struct rb_event_info *info,
3339 			     unsigned long tail)
3340 {
3341 	struct ring_buffer_event *event;
3342 	u64 ts, delta;
3343 	int e;
3344 
3345 	ts = bpage->time_stamp;
3346 	pr_warn("  [%lld] PAGE TIME STAMP\n", ts);
3347 
3348 	for (e = 0; e < tail; e += rb_event_length(event)) {
3349 
3350 		event = (struct ring_buffer_event *)(bpage->data + e);
3351 
3352 		switch (event->type_len) {
3353 
3354 		case RINGBUF_TYPE_TIME_EXTEND:
3355 			delta = rb_event_time_stamp(event);
3356 			ts += delta;
3357 			pr_warn(" 0x%x: [%lld] delta:%lld TIME EXTEND\n",
3358 				e, ts, delta);
3359 			break;
3360 
3361 		case RINGBUF_TYPE_TIME_STAMP:
3362 			delta = rb_event_time_stamp(event);
3363 			ts = rb_fix_abs_ts(delta, ts);
3364 			pr_warn(" 0x%x:  [%lld] absolute:%lld TIME STAMP\n",
3365 				e, ts, delta);
3366 			break;
3367 
3368 		case RINGBUF_TYPE_PADDING:
3369 			ts += event->time_delta;
3370 			pr_warn(" 0x%x:  [%lld] delta:%d PADDING\n",
3371 				e, ts, event->time_delta);
3372 			break;
3373 
3374 		case RINGBUF_TYPE_DATA:
3375 			ts += event->time_delta;
3376 			pr_warn(" 0x%x:  [%lld] delta:%d %s%s\n",
3377 				e, ts, event->time_delta,
3378 				show_flags(event), show_irq(event));
3379 			break;
3380 
3381 		default:
3382 			break;
3383 		}
3384 	}
3385 	pr_warn("expected end:0x%lx last event actually ended at:0x%x\n", tail, e);
3386 }
3387 
3388 static DEFINE_PER_CPU(atomic_t, checking);
3389 static atomic_t ts_dump;
3390 
3391 #define buffer_warn_return(fmt, ...)					\
3392 	do {								\
3393 		/* If another report is happening, ignore this one */	\
3394 		if (atomic_inc_return(&ts_dump) != 1) {			\
3395 			atomic_dec(&ts_dump);				\
3396 			goto out;					\
3397 		}							\
3398 		atomic_inc(&cpu_buffer->record_disabled);		\
3399 		pr_warn(fmt, ##__VA_ARGS__);				\
3400 		dump_buffer_page(bpage, info, tail);			\
3401 		atomic_dec(&ts_dump);					\
3402 		/* There's some cases in boot up that this can happen */ \
3403 		if (WARN_ON_ONCE(system_state != SYSTEM_BOOTING))	\
3404 			/* Do not re-enable checking */			\
3405 			return;						\
3406 	} while (0)
3407 
3408 /*
3409  * Check if the current event time stamp matches the deltas on
3410  * the buffer page.
3411  */
3412 static void check_buffer(struct ring_buffer_per_cpu *cpu_buffer,
3413 			 struct rb_event_info *info,
3414 			 unsigned long tail)
3415 {
3416 	struct ring_buffer_event *event;
3417 	struct buffer_data_page *bpage;
3418 	u64 ts, delta;
3419 	bool full = false;
3420 	int e;
3421 
3422 	bpage = info->tail_page->page;
3423 
3424 	if (tail == CHECK_FULL_PAGE) {
3425 		full = true;
3426 		tail = local_read(&bpage->commit);
3427 	} else if (info->add_timestamp &
3428 		   (RB_ADD_STAMP_FORCE | RB_ADD_STAMP_ABSOLUTE)) {
3429 		/* Ignore events with absolute time stamps */
3430 		return;
3431 	}
3432 
3433 	/*
3434 	 * Do not check the first event (skip possible extends too).
3435 	 * Also do not check if previous events have not been committed.
3436 	 */
3437 	if (tail <= 8 || tail > local_read(&bpage->commit))
3438 		return;
3439 
3440 	/*
3441 	 * If this interrupted another event,
3442 	 */
3443 	if (atomic_inc_return(this_cpu_ptr(&checking)) != 1)
3444 		goto out;
3445 
3446 	ts = bpage->time_stamp;
3447 
3448 	for (e = 0; e < tail; e += rb_event_length(event)) {
3449 
3450 		event = (struct ring_buffer_event *)(bpage->data + e);
3451 
3452 		switch (event->type_len) {
3453 
3454 		case RINGBUF_TYPE_TIME_EXTEND:
3455 			delta = rb_event_time_stamp(event);
3456 			ts += delta;
3457 			break;
3458 
3459 		case RINGBUF_TYPE_TIME_STAMP:
3460 			delta = rb_event_time_stamp(event);
3461 			delta = rb_fix_abs_ts(delta, ts);
3462 			if (delta < ts) {
3463 				buffer_warn_return("[CPU: %d]ABSOLUTE TIME WENT BACKWARDS: last ts: %lld absolute ts: %lld\n",
3464 						   cpu_buffer->cpu, ts, delta);
3465 			}
3466 			ts = delta;
3467 			break;
3468 
3469 		case RINGBUF_TYPE_PADDING:
3470 			if (event->time_delta == 1)
3471 				break;
3472 			fallthrough;
3473 		case RINGBUF_TYPE_DATA:
3474 			ts += event->time_delta;
3475 			break;
3476 
3477 		default:
3478 			RB_WARN_ON(cpu_buffer, 1);
3479 		}
3480 	}
3481 	if ((full && ts > info->ts) ||
3482 	    (!full && ts + info->delta != info->ts)) {
3483 		buffer_warn_return("[CPU: %d]TIME DOES NOT MATCH expected:%lld actual:%lld delta:%lld before:%lld after:%lld%s context:%s\n",
3484 				   cpu_buffer->cpu,
3485 				   ts + info->delta, info->ts, info->delta,
3486 				   info->before, info->after,
3487 				   full ? " (full)" : "", show_interrupt_level());
3488 	}
3489 out:
3490 	atomic_dec(this_cpu_ptr(&checking));
3491 }
3492 #else
3493 static inline void check_buffer(struct ring_buffer_per_cpu *cpu_buffer,
3494 			 struct rb_event_info *info,
3495 			 unsigned long tail)
3496 {
3497 }
3498 #endif /* CONFIG_RING_BUFFER_VALIDATE_TIME_DELTAS */
3499 
3500 static struct ring_buffer_event *
3501 __rb_reserve_next(struct ring_buffer_per_cpu *cpu_buffer,
3502 		  struct rb_event_info *info)
3503 {
3504 	struct ring_buffer_event *event;
3505 	struct buffer_page *tail_page;
3506 	unsigned long tail, write, w;
3507 
3508 	/* Don't let the compiler play games with cpu_buffer->tail_page */
3509 	tail_page = info->tail_page = READ_ONCE(cpu_buffer->tail_page);
3510 
3511  /*A*/	w = local_read(&tail_page->write) & RB_WRITE_MASK;
3512 	barrier();
3513 	rb_time_read(&cpu_buffer->before_stamp, &info->before);
3514 	rb_time_read(&cpu_buffer->write_stamp, &info->after);
3515 	barrier();
3516 	info->ts = rb_time_stamp(cpu_buffer->buffer);
3517 
3518 	if ((info->add_timestamp & RB_ADD_STAMP_ABSOLUTE)) {
3519 		info->delta = info->ts;
3520 	} else {
3521 		/*
3522 		 * If interrupting an event time update, we may need an
3523 		 * absolute timestamp.
3524 		 * Don't bother if this is the start of a new page (w == 0).
3525 		 */
3526 		if (!w) {
3527 			/* Use the sub-buffer timestamp */
3528 			info->delta = 0;
3529 		} else if (unlikely(info->before != info->after)) {
3530 			info->add_timestamp |= RB_ADD_STAMP_FORCE | RB_ADD_STAMP_EXTEND;
3531 			info->length += RB_LEN_TIME_EXTEND;
3532 		} else {
3533 			info->delta = info->ts - info->after;
3534 			if (unlikely(test_time_stamp(info->delta))) {
3535 				info->add_timestamp |= RB_ADD_STAMP_EXTEND;
3536 				info->length += RB_LEN_TIME_EXTEND;
3537 			}
3538 		}
3539 	}
3540 
3541  /*B*/	rb_time_set(&cpu_buffer->before_stamp, info->ts);
3542 
3543  /*C*/	write = local_add_return(info->length, &tail_page->write);
3544 
3545 	/* set write to only the index of the write */
3546 	write &= RB_WRITE_MASK;
3547 
3548 	tail = write - info->length;
3549 
3550 	/* See if we shot pass the end of this buffer page */
3551 	if (unlikely(write > cpu_buffer->buffer->subbuf_size)) {
3552 		check_buffer(cpu_buffer, info, CHECK_FULL_PAGE);
3553 		return rb_move_tail(cpu_buffer, tail, info);
3554 	}
3555 
3556 	if (likely(tail == w)) {
3557 		/* Nothing interrupted us between A and C */
3558  /*D*/		rb_time_set(&cpu_buffer->write_stamp, info->ts);
3559 		/*
3560 		 * If something came in between C and D, the write stamp
3561 		 * may now not be in sync. But that's fine as the before_stamp
3562 		 * will be different and then next event will just be forced
3563 		 * to use an absolute timestamp.
3564 		 */
3565 		if (likely(!(info->add_timestamp &
3566 			     (RB_ADD_STAMP_FORCE | RB_ADD_STAMP_ABSOLUTE))))
3567 			/* This did not interrupt any time update */
3568 			info->delta = info->ts - info->after;
3569 		else
3570 			/* Just use full timestamp for interrupting event */
3571 			info->delta = info->ts;
3572 		check_buffer(cpu_buffer, info, tail);
3573 	} else {
3574 		u64 ts;
3575 		/* SLOW PATH - Interrupted between A and C */
3576 
3577 		/* Save the old before_stamp */
3578 		rb_time_read(&cpu_buffer->before_stamp, &info->before);
3579 
3580 		/*
3581 		 * Read a new timestamp and update the before_stamp to make
3582 		 * the next event after this one force using an absolute
3583 		 * timestamp. This is in case an interrupt were to come in
3584 		 * between E and F.
3585 		 */
3586 		ts = rb_time_stamp(cpu_buffer->buffer);
3587 		rb_time_set(&cpu_buffer->before_stamp, ts);
3588 
3589 		barrier();
3590  /*E*/		rb_time_read(&cpu_buffer->write_stamp, &info->after);
3591 		barrier();
3592  /*F*/		if (write == (local_read(&tail_page->write) & RB_WRITE_MASK) &&
3593 		    info->after == info->before && info->after < ts) {
3594 			/*
3595 			 * Nothing came after this event between C and F, it is
3596 			 * safe to use info->after for the delta as it
3597 			 * matched info->before and is still valid.
3598 			 */
3599 			info->delta = ts - info->after;
3600 		} else {
3601 			/*
3602 			 * Interrupted between C and F:
3603 			 * Lost the previous events time stamp. Just set the
3604 			 * delta to zero, and this will be the same time as
3605 			 * the event this event interrupted. And the events that
3606 			 * came after this will still be correct (as they would
3607 			 * have built their delta on the previous event.
3608 			 */
3609 			info->delta = 0;
3610 		}
3611 		info->ts = ts;
3612 		info->add_timestamp &= ~RB_ADD_STAMP_FORCE;
3613 	}
3614 
3615 	/*
3616 	 * If this is the first commit on the page, then it has the same
3617 	 * timestamp as the page itself.
3618 	 */
3619 	if (unlikely(!tail && !(info->add_timestamp &
3620 				(RB_ADD_STAMP_FORCE | RB_ADD_STAMP_ABSOLUTE))))
3621 		info->delta = 0;
3622 
3623 	/* We reserved something on the buffer */
3624 
3625 	event = __rb_page_index(tail_page, tail);
3626 	rb_update_event(cpu_buffer, event, info);
3627 
3628 	local_inc(&tail_page->entries);
3629 
3630 	/*
3631 	 * If this is the first commit on the page, then update
3632 	 * its timestamp.
3633 	 */
3634 	if (unlikely(!tail))
3635 		tail_page->page->time_stamp = info->ts;
3636 
3637 	/* account for these added bytes */
3638 	local_add(info->length, &cpu_buffer->entries_bytes);
3639 
3640 	return event;
3641 }
3642 
3643 static __always_inline struct ring_buffer_event *
3644 rb_reserve_next_event(struct trace_buffer *buffer,
3645 		      struct ring_buffer_per_cpu *cpu_buffer,
3646 		      unsigned long length)
3647 {
3648 	struct ring_buffer_event *event;
3649 	struct rb_event_info info;
3650 	int nr_loops = 0;
3651 	int add_ts_default;
3652 
3653 	/* ring buffer does cmpxchg, make sure it is safe in NMI context */
3654 	if (!IS_ENABLED(CONFIG_ARCH_HAVE_NMI_SAFE_CMPXCHG) &&
3655 	    (unlikely(in_nmi()))) {
3656 		return NULL;
3657 	}
3658 
3659 	rb_start_commit(cpu_buffer);
3660 	/* The commit page can not change after this */
3661 
3662 #ifdef CONFIG_RING_BUFFER_ALLOW_SWAP
3663 	/*
3664 	 * Due to the ability to swap a cpu buffer from a buffer
3665 	 * it is possible it was swapped before we committed.
3666 	 * (committing stops a swap). We check for it here and
3667 	 * if it happened, we have to fail the write.
3668 	 */
3669 	barrier();
3670 	if (unlikely(READ_ONCE(cpu_buffer->buffer) != buffer)) {
3671 		local_dec(&cpu_buffer->committing);
3672 		local_dec(&cpu_buffer->commits);
3673 		return NULL;
3674 	}
3675 #endif
3676 
3677 	info.length = rb_calculate_event_length(length);
3678 
3679 	if (ring_buffer_time_stamp_abs(cpu_buffer->buffer)) {
3680 		add_ts_default = RB_ADD_STAMP_ABSOLUTE;
3681 		info.length += RB_LEN_TIME_EXTEND;
3682 		if (info.length > cpu_buffer->buffer->max_data_size)
3683 			goto out_fail;
3684 	} else {
3685 		add_ts_default = RB_ADD_STAMP_NONE;
3686 	}
3687 
3688  again:
3689 	info.add_timestamp = add_ts_default;
3690 	info.delta = 0;
3691 
3692 	/*
3693 	 * We allow for interrupts to reenter here and do a trace.
3694 	 * If one does, it will cause this original code to loop
3695 	 * back here. Even with heavy interrupts happening, this
3696 	 * should only happen a few times in a row. If this happens
3697 	 * 1000 times in a row, there must be either an interrupt
3698 	 * storm or we have something buggy.
3699 	 * Bail!
3700 	 */
3701 	if (RB_WARN_ON(cpu_buffer, ++nr_loops > 1000))
3702 		goto out_fail;
3703 
3704 	event = __rb_reserve_next(cpu_buffer, &info);
3705 
3706 	if (unlikely(PTR_ERR(event) == -EAGAIN)) {
3707 		if (info.add_timestamp & (RB_ADD_STAMP_FORCE | RB_ADD_STAMP_EXTEND))
3708 			info.length -= RB_LEN_TIME_EXTEND;
3709 		goto again;
3710 	}
3711 
3712 	if (likely(event))
3713 		return event;
3714  out_fail:
3715 	rb_end_commit(cpu_buffer);
3716 	return NULL;
3717 }
3718 
3719 /**
3720  * ring_buffer_lock_reserve - reserve a part of the buffer
3721  * @buffer: the ring buffer to reserve from
3722  * @length: the length of the data to reserve (excluding event header)
3723  *
3724  * Returns a reserved event on the ring buffer to copy directly to.
3725  * The user of this interface will need to get the body to write into
3726  * and can use the ring_buffer_event_data() interface.
3727  *
3728  * The length is the length of the data needed, not the event length
3729  * which also includes the event header.
3730  *
3731  * Must be paired with ring_buffer_unlock_commit, unless NULL is returned.
3732  * If NULL is returned, then nothing has been allocated or locked.
3733  */
3734 struct ring_buffer_event *
3735 ring_buffer_lock_reserve(struct trace_buffer *buffer, unsigned long length)
3736 {
3737 	struct ring_buffer_per_cpu *cpu_buffer;
3738 	struct ring_buffer_event *event;
3739 	int cpu;
3740 
3741 	/* If we are tracing schedule, we don't want to recurse */
3742 	preempt_disable_notrace();
3743 
3744 	if (unlikely(atomic_read(&buffer->record_disabled)))
3745 		goto out;
3746 
3747 	cpu = raw_smp_processor_id();
3748 
3749 	if (unlikely(!cpumask_test_cpu(cpu, buffer->cpumask)))
3750 		goto out;
3751 
3752 	cpu_buffer = buffer->buffers[cpu];
3753 
3754 	if (unlikely(atomic_read(&cpu_buffer->record_disabled)))
3755 		goto out;
3756 
3757 	if (unlikely(length > buffer->max_data_size))
3758 		goto out;
3759 
3760 	if (unlikely(trace_recursive_lock(cpu_buffer)))
3761 		goto out;
3762 
3763 	event = rb_reserve_next_event(buffer, cpu_buffer, length);
3764 	if (!event)
3765 		goto out_unlock;
3766 
3767 	return event;
3768 
3769  out_unlock:
3770 	trace_recursive_unlock(cpu_buffer);
3771  out:
3772 	preempt_enable_notrace();
3773 	return NULL;
3774 }
3775 EXPORT_SYMBOL_GPL(ring_buffer_lock_reserve);
3776 
3777 /*
3778  * Decrement the entries to the page that an event is on.
3779  * The event does not even need to exist, only the pointer
3780  * to the page it is on. This may only be called before the commit
3781  * takes place.
3782  */
3783 static inline void
3784 rb_decrement_entry(struct ring_buffer_per_cpu *cpu_buffer,
3785 		   struct ring_buffer_event *event)
3786 {
3787 	unsigned long addr = (unsigned long)event;
3788 	struct buffer_page *bpage = cpu_buffer->commit_page;
3789 	struct buffer_page *start;
3790 
3791 	addr &= ~((PAGE_SIZE << cpu_buffer->buffer->subbuf_order) - 1);
3792 
3793 	/* Do the likely case first */
3794 	if (likely(bpage->page == (void *)addr)) {
3795 		local_dec(&bpage->entries);
3796 		return;
3797 	}
3798 
3799 	/*
3800 	 * Because the commit page may be on the reader page we
3801 	 * start with the next page and check the end loop there.
3802 	 */
3803 	rb_inc_page(&bpage);
3804 	start = bpage;
3805 	do {
3806 		if (bpage->page == (void *)addr) {
3807 			local_dec(&bpage->entries);
3808 			return;
3809 		}
3810 		rb_inc_page(&bpage);
3811 	} while (bpage != start);
3812 
3813 	/* commit not part of this buffer?? */
3814 	RB_WARN_ON(cpu_buffer, 1);
3815 }
3816 
3817 /**
3818  * ring_buffer_discard_commit - discard an event that has not been committed
3819  * @buffer: the ring buffer
3820  * @event: non committed event to discard
3821  *
3822  * Sometimes an event that is in the ring buffer needs to be ignored.
3823  * This function lets the user discard an event in the ring buffer
3824  * and then that event will not be read later.
3825  *
3826  * This function only works if it is called before the item has been
3827  * committed. It will try to free the event from the ring buffer
3828  * if another event has not been added behind it.
3829  *
3830  * If another event has been added behind it, it will set the event
3831  * up as discarded, and perform the commit.
3832  *
3833  * If this function is called, do not call ring_buffer_unlock_commit on
3834  * the event.
3835  */
3836 void ring_buffer_discard_commit(struct trace_buffer *buffer,
3837 				struct ring_buffer_event *event)
3838 {
3839 	struct ring_buffer_per_cpu *cpu_buffer;
3840 	int cpu;
3841 
3842 	/* The event is discarded regardless */
3843 	rb_event_discard(event);
3844 
3845 	cpu = smp_processor_id();
3846 	cpu_buffer = buffer->buffers[cpu];
3847 
3848 	/*
3849 	 * This must only be called if the event has not been
3850 	 * committed yet. Thus we can assume that preemption
3851 	 * is still disabled.
3852 	 */
3853 	RB_WARN_ON(buffer, !local_read(&cpu_buffer->committing));
3854 
3855 	rb_decrement_entry(cpu_buffer, event);
3856 	if (rb_try_to_discard(cpu_buffer, event))
3857 		goto out;
3858 
3859  out:
3860 	rb_end_commit(cpu_buffer);
3861 
3862 	trace_recursive_unlock(cpu_buffer);
3863 
3864 	preempt_enable_notrace();
3865 
3866 }
3867 EXPORT_SYMBOL_GPL(ring_buffer_discard_commit);
3868 
3869 /**
3870  * ring_buffer_write - write data to the buffer without reserving
3871  * @buffer: The ring buffer to write to.
3872  * @length: The length of the data being written (excluding the event header)
3873  * @data: The data to write to the buffer.
3874  *
3875  * This is like ring_buffer_lock_reserve and ring_buffer_unlock_commit as
3876  * one function. If you already have the data to write to the buffer, it
3877  * may be easier to simply call this function.
3878  *
3879  * Note, like ring_buffer_lock_reserve, the length is the length of the data
3880  * and not the length of the event which would hold the header.
3881  */
3882 int ring_buffer_write(struct trace_buffer *buffer,
3883 		      unsigned long length,
3884 		      void *data)
3885 {
3886 	struct ring_buffer_per_cpu *cpu_buffer;
3887 	struct ring_buffer_event *event;
3888 	void *body;
3889 	int ret = -EBUSY;
3890 	int cpu;
3891 
3892 	preempt_disable_notrace();
3893 
3894 	if (atomic_read(&buffer->record_disabled))
3895 		goto out;
3896 
3897 	cpu = raw_smp_processor_id();
3898 
3899 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
3900 		goto out;
3901 
3902 	cpu_buffer = buffer->buffers[cpu];
3903 
3904 	if (atomic_read(&cpu_buffer->record_disabled))
3905 		goto out;
3906 
3907 	if (length > buffer->max_data_size)
3908 		goto out;
3909 
3910 	if (unlikely(trace_recursive_lock(cpu_buffer)))
3911 		goto out;
3912 
3913 	event = rb_reserve_next_event(buffer, cpu_buffer, length);
3914 	if (!event)
3915 		goto out_unlock;
3916 
3917 	body = rb_event_data(event);
3918 
3919 	memcpy(body, data, length);
3920 
3921 	rb_commit(cpu_buffer);
3922 
3923 	rb_wakeups(buffer, cpu_buffer);
3924 
3925 	ret = 0;
3926 
3927  out_unlock:
3928 	trace_recursive_unlock(cpu_buffer);
3929 
3930  out:
3931 	preempt_enable_notrace();
3932 
3933 	return ret;
3934 }
3935 EXPORT_SYMBOL_GPL(ring_buffer_write);
3936 
3937 static bool rb_per_cpu_empty(struct ring_buffer_per_cpu *cpu_buffer)
3938 {
3939 	struct buffer_page *reader = cpu_buffer->reader_page;
3940 	struct buffer_page *head = rb_set_head_page(cpu_buffer);
3941 	struct buffer_page *commit = cpu_buffer->commit_page;
3942 
3943 	/* In case of error, head will be NULL */
3944 	if (unlikely(!head))
3945 		return true;
3946 
3947 	/* Reader should exhaust content in reader page */
3948 	if (reader->read != rb_page_commit(reader))
3949 		return false;
3950 
3951 	/*
3952 	 * If writers are committing on the reader page, knowing all
3953 	 * committed content has been read, the ring buffer is empty.
3954 	 */
3955 	if (commit == reader)
3956 		return true;
3957 
3958 	/*
3959 	 * If writers are committing on a page other than reader page
3960 	 * and head page, there should always be content to read.
3961 	 */
3962 	if (commit != head)
3963 		return false;
3964 
3965 	/*
3966 	 * Writers are committing on the head page, we just need
3967 	 * to care about there're committed data, and the reader will
3968 	 * swap reader page with head page when it is to read data.
3969 	 */
3970 	return rb_page_commit(commit) == 0;
3971 }
3972 
3973 /**
3974  * ring_buffer_record_disable - stop all writes into the buffer
3975  * @buffer: The ring buffer to stop writes to.
3976  *
3977  * This prevents all writes to the buffer. Any attempt to write
3978  * to the buffer after this will fail and return NULL.
3979  *
3980  * The caller should call synchronize_rcu() after this.
3981  */
3982 void ring_buffer_record_disable(struct trace_buffer *buffer)
3983 {
3984 	atomic_inc(&buffer->record_disabled);
3985 }
3986 EXPORT_SYMBOL_GPL(ring_buffer_record_disable);
3987 
3988 /**
3989  * ring_buffer_record_enable - enable writes to the buffer
3990  * @buffer: The ring buffer to enable writes
3991  *
3992  * Note, multiple disables will need the same number of enables
3993  * to truly enable the writing (much like preempt_disable).
3994  */
3995 void ring_buffer_record_enable(struct trace_buffer *buffer)
3996 {
3997 	atomic_dec(&buffer->record_disabled);
3998 }
3999 EXPORT_SYMBOL_GPL(ring_buffer_record_enable);
4000 
4001 /**
4002  * ring_buffer_record_off - stop all writes into the buffer
4003  * @buffer: The ring buffer to stop writes to.
4004  *
4005  * This prevents all writes to the buffer. Any attempt to write
4006  * to the buffer after this will fail and return NULL.
4007  *
4008  * This is different than ring_buffer_record_disable() as
4009  * it works like an on/off switch, where as the disable() version
4010  * must be paired with a enable().
4011  */
4012 void ring_buffer_record_off(struct trace_buffer *buffer)
4013 {
4014 	unsigned int rd;
4015 	unsigned int new_rd;
4016 
4017 	rd = atomic_read(&buffer->record_disabled);
4018 	do {
4019 		new_rd = rd | RB_BUFFER_OFF;
4020 	} while (!atomic_try_cmpxchg(&buffer->record_disabled, &rd, new_rd));
4021 }
4022 EXPORT_SYMBOL_GPL(ring_buffer_record_off);
4023 
4024 /**
4025  * ring_buffer_record_on - restart writes into the buffer
4026  * @buffer: The ring buffer to start writes to.
4027  *
4028  * This enables all writes to the buffer that was disabled by
4029  * ring_buffer_record_off().
4030  *
4031  * This is different than ring_buffer_record_enable() as
4032  * it works like an on/off switch, where as the enable() version
4033  * must be paired with a disable().
4034  */
4035 void ring_buffer_record_on(struct trace_buffer *buffer)
4036 {
4037 	unsigned int rd;
4038 	unsigned int new_rd;
4039 
4040 	rd = atomic_read(&buffer->record_disabled);
4041 	do {
4042 		new_rd = rd & ~RB_BUFFER_OFF;
4043 	} while (!atomic_try_cmpxchg(&buffer->record_disabled, &rd, new_rd));
4044 }
4045 EXPORT_SYMBOL_GPL(ring_buffer_record_on);
4046 
4047 /**
4048  * ring_buffer_record_is_on - return true if the ring buffer can write
4049  * @buffer: The ring buffer to see if write is enabled
4050  *
4051  * Returns true if the ring buffer is in a state that it accepts writes.
4052  */
4053 bool ring_buffer_record_is_on(struct trace_buffer *buffer)
4054 {
4055 	return !atomic_read(&buffer->record_disabled);
4056 }
4057 
4058 /**
4059  * ring_buffer_record_is_set_on - return true if the ring buffer is set writable
4060  * @buffer: The ring buffer to see if write is set enabled
4061  *
4062  * Returns true if the ring buffer is set writable by ring_buffer_record_on().
4063  * Note that this does NOT mean it is in a writable state.
4064  *
4065  * It may return true when the ring buffer has been disabled by
4066  * ring_buffer_record_disable(), as that is a temporary disabling of
4067  * the ring buffer.
4068  */
4069 bool ring_buffer_record_is_set_on(struct trace_buffer *buffer)
4070 {
4071 	return !(atomic_read(&buffer->record_disabled) & RB_BUFFER_OFF);
4072 }
4073 
4074 /**
4075  * ring_buffer_record_disable_cpu - stop all writes into the cpu_buffer
4076  * @buffer: The ring buffer to stop writes to.
4077  * @cpu: The CPU buffer to stop
4078  *
4079  * This prevents all writes to the buffer. Any attempt to write
4080  * to the buffer after this will fail and return NULL.
4081  *
4082  * The caller should call synchronize_rcu() after this.
4083  */
4084 void ring_buffer_record_disable_cpu(struct trace_buffer *buffer, int cpu)
4085 {
4086 	struct ring_buffer_per_cpu *cpu_buffer;
4087 
4088 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
4089 		return;
4090 
4091 	cpu_buffer = buffer->buffers[cpu];
4092 	atomic_inc(&cpu_buffer->record_disabled);
4093 }
4094 EXPORT_SYMBOL_GPL(ring_buffer_record_disable_cpu);
4095 
4096 /**
4097  * ring_buffer_record_enable_cpu - enable writes to the buffer
4098  * @buffer: The ring buffer to enable writes
4099  * @cpu: The CPU to enable.
4100  *
4101  * Note, multiple disables will need the same number of enables
4102  * to truly enable the writing (much like preempt_disable).
4103  */
4104 void ring_buffer_record_enable_cpu(struct trace_buffer *buffer, int cpu)
4105 {
4106 	struct ring_buffer_per_cpu *cpu_buffer;
4107 
4108 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
4109 		return;
4110 
4111 	cpu_buffer = buffer->buffers[cpu];
4112 	atomic_dec(&cpu_buffer->record_disabled);
4113 }
4114 EXPORT_SYMBOL_GPL(ring_buffer_record_enable_cpu);
4115 
4116 /*
4117  * The total entries in the ring buffer is the running counter
4118  * of entries entered into the ring buffer, minus the sum of
4119  * the entries read from the ring buffer and the number of
4120  * entries that were overwritten.
4121  */
4122 static inline unsigned long
4123 rb_num_of_entries(struct ring_buffer_per_cpu *cpu_buffer)
4124 {
4125 	return local_read(&cpu_buffer->entries) -
4126 		(local_read(&cpu_buffer->overrun) + cpu_buffer->read);
4127 }
4128 
4129 /**
4130  * ring_buffer_oldest_event_ts - get the oldest event timestamp from the buffer
4131  * @buffer: The ring buffer
4132  * @cpu: The per CPU buffer to read from.
4133  */
4134 u64 ring_buffer_oldest_event_ts(struct trace_buffer *buffer, int cpu)
4135 {
4136 	unsigned long flags;
4137 	struct ring_buffer_per_cpu *cpu_buffer;
4138 	struct buffer_page *bpage;
4139 	u64 ret = 0;
4140 
4141 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
4142 		return 0;
4143 
4144 	cpu_buffer = buffer->buffers[cpu];
4145 	raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
4146 	/*
4147 	 * if the tail is on reader_page, oldest time stamp is on the reader
4148 	 * page
4149 	 */
4150 	if (cpu_buffer->tail_page == cpu_buffer->reader_page)
4151 		bpage = cpu_buffer->reader_page;
4152 	else
4153 		bpage = rb_set_head_page(cpu_buffer);
4154 	if (bpage)
4155 		ret = bpage->page->time_stamp;
4156 	raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
4157 
4158 	return ret;
4159 }
4160 EXPORT_SYMBOL_GPL(ring_buffer_oldest_event_ts);
4161 
4162 /**
4163  * ring_buffer_bytes_cpu - get the number of bytes unconsumed in a cpu buffer
4164  * @buffer: The ring buffer
4165  * @cpu: The per CPU buffer to read from.
4166  */
4167 unsigned long ring_buffer_bytes_cpu(struct trace_buffer *buffer, int cpu)
4168 {
4169 	struct ring_buffer_per_cpu *cpu_buffer;
4170 	unsigned long ret;
4171 
4172 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
4173 		return 0;
4174 
4175 	cpu_buffer = buffer->buffers[cpu];
4176 	ret = local_read(&cpu_buffer->entries_bytes) - cpu_buffer->read_bytes;
4177 
4178 	return ret;
4179 }
4180 EXPORT_SYMBOL_GPL(ring_buffer_bytes_cpu);
4181 
4182 /**
4183  * ring_buffer_entries_cpu - get the number of entries in a cpu buffer
4184  * @buffer: The ring buffer
4185  * @cpu: The per CPU buffer to get the entries from.
4186  */
4187 unsigned long ring_buffer_entries_cpu(struct trace_buffer *buffer, int cpu)
4188 {
4189 	struct ring_buffer_per_cpu *cpu_buffer;
4190 
4191 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
4192 		return 0;
4193 
4194 	cpu_buffer = buffer->buffers[cpu];
4195 
4196 	return rb_num_of_entries(cpu_buffer);
4197 }
4198 EXPORT_SYMBOL_GPL(ring_buffer_entries_cpu);
4199 
4200 /**
4201  * ring_buffer_overrun_cpu - get the number of overruns caused by the ring
4202  * buffer wrapping around (only if RB_FL_OVERWRITE is on).
4203  * @buffer: The ring buffer
4204  * @cpu: The per CPU buffer to get the number of overruns from
4205  */
4206 unsigned long ring_buffer_overrun_cpu(struct trace_buffer *buffer, int cpu)
4207 {
4208 	struct ring_buffer_per_cpu *cpu_buffer;
4209 	unsigned long ret;
4210 
4211 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
4212 		return 0;
4213 
4214 	cpu_buffer = buffer->buffers[cpu];
4215 	ret = local_read(&cpu_buffer->overrun);
4216 
4217 	return ret;
4218 }
4219 EXPORT_SYMBOL_GPL(ring_buffer_overrun_cpu);
4220 
4221 /**
4222  * ring_buffer_commit_overrun_cpu - get the number of overruns caused by
4223  * commits failing due to the buffer wrapping around while there are uncommitted
4224  * events, such as during an interrupt storm.
4225  * @buffer: The ring buffer
4226  * @cpu: The per CPU buffer to get the number of overruns from
4227  */
4228 unsigned long
4229 ring_buffer_commit_overrun_cpu(struct trace_buffer *buffer, int cpu)
4230 {
4231 	struct ring_buffer_per_cpu *cpu_buffer;
4232 	unsigned long ret;
4233 
4234 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
4235 		return 0;
4236 
4237 	cpu_buffer = buffer->buffers[cpu];
4238 	ret = local_read(&cpu_buffer->commit_overrun);
4239 
4240 	return ret;
4241 }
4242 EXPORT_SYMBOL_GPL(ring_buffer_commit_overrun_cpu);
4243 
4244 /**
4245  * ring_buffer_dropped_events_cpu - get the number of dropped events caused by
4246  * the ring buffer filling up (only if RB_FL_OVERWRITE is off).
4247  * @buffer: The ring buffer
4248  * @cpu: The per CPU buffer to get the number of overruns from
4249  */
4250 unsigned long
4251 ring_buffer_dropped_events_cpu(struct trace_buffer *buffer, int cpu)
4252 {
4253 	struct ring_buffer_per_cpu *cpu_buffer;
4254 	unsigned long ret;
4255 
4256 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
4257 		return 0;
4258 
4259 	cpu_buffer = buffer->buffers[cpu];
4260 	ret = local_read(&cpu_buffer->dropped_events);
4261 
4262 	return ret;
4263 }
4264 EXPORT_SYMBOL_GPL(ring_buffer_dropped_events_cpu);
4265 
4266 /**
4267  * ring_buffer_read_events_cpu - get the number of events successfully read
4268  * @buffer: The ring buffer
4269  * @cpu: The per CPU buffer to get the number of events read
4270  */
4271 unsigned long
4272 ring_buffer_read_events_cpu(struct trace_buffer *buffer, int cpu)
4273 {
4274 	struct ring_buffer_per_cpu *cpu_buffer;
4275 
4276 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
4277 		return 0;
4278 
4279 	cpu_buffer = buffer->buffers[cpu];
4280 	return cpu_buffer->read;
4281 }
4282 EXPORT_SYMBOL_GPL(ring_buffer_read_events_cpu);
4283 
4284 /**
4285  * ring_buffer_entries - get the number of entries in a buffer
4286  * @buffer: The ring buffer
4287  *
4288  * Returns the total number of entries in the ring buffer
4289  * (all CPU entries)
4290  */
4291 unsigned long ring_buffer_entries(struct trace_buffer *buffer)
4292 {
4293 	struct ring_buffer_per_cpu *cpu_buffer;
4294 	unsigned long entries = 0;
4295 	int cpu;
4296 
4297 	/* if you care about this being correct, lock the buffer */
4298 	for_each_buffer_cpu(buffer, cpu) {
4299 		cpu_buffer = buffer->buffers[cpu];
4300 		entries += rb_num_of_entries(cpu_buffer);
4301 	}
4302 
4303 	return entries;
4304 }
4305 EXPORT_SYMBOL_GPL(ring_buffer_entries);
4306 
4307 /**
4308  * ring_buffer_overruns - get the number of overruns in buffer
4309  * @buffer: The ring buffer
4310  *
4311  * Returns the total number of overruns in the ring buffer
4312  * (all CPU entries)
4313  */
4314 unsigned long ring_buffer_overruns(struct trace_buffer *buffer)
4315 {
4316 	struct ring_buffer_per_cpu *cpu_buffer;
4317 	unsigned long overruns = 0;
4318 	int cpu;
4319 
4320 	/* if you care about this being correct, lock the buffer */
4321 	for_each_buffer_cpu(buffer, cpu) {
4322 		cpu_buffer = buffer->buffers[cpu];
4323 		overruns += local_read(&cpu_buffer->overrun);
4324 	}
4325 
4326 	return overruns;
4327 }
4328 EXPORT_SYMBOL_GPL(ring_buffer_overruns);
4329 
4330 static void rb_iter_reset(struct ring_buffer_iter *iter)
4331 {
4332 	struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
4333 
4334 	/* Iterator usage is expected to have record disabled */
4335 	iter->head_page = cpu_buffer->reader_page;
4336 	iter->head = cpu_buffer->reader_page->read;
4337 	iter->next_event = iter->head;
4338 
4339 	iter->cache_reader_page = iter->head_page;
4340 	iter->cache_read = cpu_buffer->read;
4341 	iter->cache_pages_removed = cpu_buffer->pages_removed;
4342 
4343 	if (iter->head) {
4344 		iter->read_stamp = cpu_buffer->read_stamp;
4345 		iter->page_stamp = cpu_buffer->reader_page->page->time_stamp;
4346 	} else {
4347 		iter->read_stamp = iter->head_page->page->time_stamp;
4348 		iter->page_stamp = iter->read_stamp;
4349 	}
4350 }
4351 
4352 /**
4353  * ring_buffer_iter_reset - reset an iterator
4354  * @iter: The iterator to reset
4355  *
4356  * Resets the iterator, so that it will start from the beginning
4357  * again.
4358  */
4359 void ring_buffer_iter_reset(struct ring_buffer_iter *iter)
4360 {
4361 	struct ring_buffer_per_cpu *cpu_buffer;
4362 	unsigned long flags;
4363 
4364 	if (!iter)
4365 		return;
4366 
4367 	cpu_buffer = iter->cpu_buffer;
4368 
4369 	raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
4370 	rb_iter_reset(iter);
4371 	raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
4372 }
4373 EXPORT_SYMBOL_GPL(ring_buffer_iter_reset);
4374 
4375 /**
4376  * ring_buffer_iter_empty - check if an iterator has no more to read
4377  * @iter: The iterator to check
4378  */
4379 int ring_buffer_iter_empty(struct ring_buffer_iter *iter)
4380 {
4381 	struct ring_buffer_per_cpu *cpu_buffer;
4382 	struct buffer_page *reader;
4383 	struct buffer_page *head_page;
4384 	struct buffer_page *commit_page;
4385 	struct buffer_page *curr_commit_page;
4386 	unsigned commit;
4387 	u64 curr_commit_ts;
4388 	u64 commit_ts;
4389 
4390 	cpu_buffer = iter->cpu_buffer;
4391 	reader = cpu_buffer->reader_page;
4392 	head_page = cpu_buffer->head_page;
4393 	commit_page = READ_ONCE(cpu_buffer->commit_page);
4394 	commit_ts = commit_page->page->time_stamp;
4395 
4396 	/*
4397 	 * When the writer goes across pages, it issues a cmpxchg which
4398 	 * is a mb(), which will synchronize with the rmb here.
4399 	 * (see rb_tail_page_update())
4400 	 */
4401 	smp_rmb();
4402 	commit = rb_page_commit(commit_page);
4403 	/* We want to make sure that the commit page doesn't change */
4404 	smp_rmb();
4405 
4406 	/* Make sure commit page didn't change */
4407 	curr_commit_page = READ_ONCE(cpu_buffer->commit_page);
4408 	curr_commit_ts = READ_ONCE(curr_commit_page->page->time_stamp);
4409 
4410 	/* If the commit page changed, then there's more data */
4411 	if (curr_commit_page != commit_page ||
4412 	    curr_commit_ts != commit_ts)
4413 		return 0;
4414 
4415 	/* Still racy, as it may return a false positive, but that's OK */
4416 	return ((iter->head_page == commit_page && iter->head >= commit) ||
4417 		(iter->head_page == reader && commit_page == head_page &&
4418 		 head_page->read == commit &&
4419 		 iter->head == rb_page_commit(cpu_buffer->reader_page)));
4420 }
4421 EXPORT_SYMBOL_GPL(ring_buffer_iter_empty);
4422 
4423 static void
4424 rb_update_read_stamp(struct ring_buffer_per_cpu *cpu_buffer,
4425 		     struct ring_buffer_event *event)
4426 {
4427 	u64 delta;
4428 
4429 	switch (event->type_len) {
4430 	case RINGBUF_TYPE_PADDING:
4431 		return;
4432 
4433 	case RINGBUF_TYPE_TIME_EXTEND:
4434 		delta = rb_event_time_stamp(event);
4435 		cpu_buffer->read_stamp += delta;
4436 		return;
4437 
4438 	case RINGBUF_TYPE_TIME_STAMP:
4439 		delta = rb_event_time_stamp(event);
4440 		delta = rb_fix_abs_ts(delta, cpu_buffer->read_stamp);
4441 		cpu_buffer->read_stamp = delta;
4442 		return;
4443 
4444 	case RINGBUF_TYPE_DATA:
4445 		cpu_buffer->read_stamp += event->time_delta;
4446 		return;
4447 
4448 	default:
4449 		RB_WARN_ON(cpu_buffer, 1);
4450 	}
4451 }
4452 
4453 static void
4454 rb_update_iter_read_stamp(struct ring_buffer_iter *iter,
4455 			  struct ring_buffer_event *event)
4456 {
4457 	u64 delta;
4458 
4459 	switch (event->type_len) {
4460 	case RINGBUF_TYPE_PADDING:
4461 		return;
4462 
4463 	case RINGBUF_TYPE_TIME_EXTEND:
4464 		delta = rb_event_time_stamp(event);
4465 		iter->read_stamp += delta;
4466 		return;
4467 
4468 	case RINGBUF_TYPE_TIME_STAMP:
4469 		delta = rb_event_time_stamp(event);
4470 		delta = rb_fix_abs_ts(delta, iter->read_stamp);
4471 		iter->read_stamp = delta;
4472 		return;
4473 
4474 	case RINGBUF_TYPE_DATA:
4475 		iter->read_stamp += event->time_delta;
4476 		return;
4477 
4478 	default:
4479 		RB_WARN_ON(iter->cpu_buffer, 1);
4480 	}
4481 }
4482 
4483 static struct buffer_page *
4484 rb_get_reader_page(struct ring_buffer_per_cpu *cpu_buffer)
4485 {
4486 	struct buffer_page *reader = NULL;
4487 	unsigned long bsize = READ_ONCE(cpu_buffer->buffer->subbuf_size);
4488 	unsigned long overwrite;
4489 	unsigned long flags;
4490 	int nr_loops = 0;
4491 	bool ret;
4492 
4493 	local_irq_save(flags);
4494 	arch_spin_lock(&cpu_buffer->lock);
4495 
4496  again:
4497 	/*
4498 	 * This should normally only loop twice. But because the
4499 	 * start of the reader inserts an empty page, it causes
4500 	 * a case where we will loop three times. There should be no
4501 	 * reason to loop four times (that I know of).
4502 	 */
4503 	if (RB_WARN_ON(cpu_buffer, ++nr_loops > 3)) {
4504 		reader = NULL;
4505 		goto out;
4506 	}
4507 
4508 	reader = cpu_buffer->reader_page;
4509 
4510 	/* If there's more to read, return this page */
4511 	if (cpu_buffer->reader_page->read < rb_page_size(reader))
4512 		goto out;
4513 
4514 	/* Never should we have an index greater than the size */
4515 	if (RB_WARN_ON(cpu_buffer,
4516 		       cpu_buffer->reader_page->read > rb_page_size(reader)))
4517 		goto out;
4518 
4519 	/* check if we caught up to the tail */
4520 	reader = NULL;
4521 	if (cpu_buffer->commit_page == cpu_buffer->reader_page)
4522 		goto out;
4523 
4524 	/* Don't bother swapping if the ring buffer is empty */
4525 	if (rb_num_of_entries(cpu_buffer) == 0)
4526 		goto out;
4527 
4528 	/*
4529 	 * Reset the reader page to size zero.
4530 	 */
4531 	local_set(&cpu_buffer->reader_page->write, 0);
4532 	local_set(&cpu_buffer->reader_page->entries, 0);
4533 	local_set(&cpu_buffer->reader_page->page->commit, 0);
4534 	cpu_buffer->reader_page->real_end = 0;
4535 
4536  spin:
4537 	/*
4538 	 * Splice the empty reader page into the list around the head.
4539 	 */
4540 	reader = rb_set_head_page(cpu_buffer);
4541 	if (!reader)
4542 		goto out;
4543 	cpu_buffer->reader_page->list.next = rb_list_head(reader->list.next);
4544 	cpu_buffer->reader_page->list.prev = reader->list.prev;
4545 
4546 	/*
4547 	 * cpu_buffer->pages just needs to point to the buffer, it
4548 	 *  has no specific buffer page to point to. Lets move it out
4549 	 *  of our way so we don't accidentally swap it.
4550 	 */
4551 	cpu_buffer->pages = reader->list.prev;
4552 
4553 	/* The reader page will be pointing to the new head */
4554 	rb_set_list_to_head(&cpu_buffer->reader_page->list);
4555 
4556 	/*
4557 	 * We want to make sure we read the overruns after we set up our
4558 	 * pointers to the next object. The writer side does a
4559 	 * cmpxchg to cross pages which acts as the mb on the writer
4560 	 * side. Note, the reader will constantly fail the swap
4561 	 * while the writer is updating the pointers, so this
4562 	 * guarantees that the overwrite recorded here is the one we
4563 	 * want to compare with the last_overrun.
4564 	 */
4565 	smp_mb();
4566 	overwrite = local_read(&(cpu_buffer->overrun));
4567 
4568 	/*
4569 	 * Here's the tricky part.
4570 	 *
4571 	 * We need to move the pointer past the header page.
4572 	 * But we can only do that if a writer is not currently
4573 	 * moving it. The page before the header page has the
4574 	 * flag bit '1' set if it is pointing to the page we want.
4575 	 * but if the writer is in the process of moving it
4576 	 * than it will be '2' or already moved '0'.
4577 	 */
4578 
4579 	ret = rb_head_page_replace(reader, cpu_buffer->reader_page);
4580 
4581 	/*
4582 	 * If we did not convert it, then we must try again.
4583 	 */
4584 	if (!ret)
4585 		goto spin;
4586 
4587 	/*
4588 	 * Yay! We succeeded in replacing the page.
4589 	 *
4590 	 * Now make the new head point back to the reader page.
4591 	 */
4592 	rb_list_head(reader->list.next)->prev = &cpu_buffer->reader_page->list;
4593 	rb_inc_page(&cpu_buffer->head_page);
4594 
4595 	local_inc(&cpu_buffer->pages_read);
4596 
4597 	/* Finally update the reader page to the new head */
4598 	cpu_buffer->reader_page = reader;
4599 	cpu_buffer->reader_page->read = 0;
4600 
4601 	if (overwrite != cpu_buffer->last_overrun) {
4602 		cpu_buffer->lost_events = overwrite - cpu_buffer->last_overrun;
4603 		cpu_buffer->last_overrun = overwrite;
4604 	}
4605 
4606 	goto again;
4607 
4608  out:
4609 	/* Update the read_stamp on the first event */
4610 	if (reader && reader->read == 0)
4611 		cpu_buffer->read_stamp = reader->page->time_stamp;
4612 
4613 	arch_spin_unlock(&cpu_buffer->lock);
4614 	local_irq_restore(flags);
4615 
4616 	/*
4617 	 * The writer has preempt disable, wait for it. But not forever
4618 	 * Although, 1 second is pretty much "forever"
4619 	 */
4620 #define USECS_WAIT	1000000
4621         for (nr_loops = 0; nr_loops < USECS_WAIT; nr_loops++) {
4622 		/* If the write is past the end of page, a writer is still updating it */
4623 		if (likely(!reader || rb_page_write(reader) <= bsize))
4624 			break;
4625 
4626 		udelay(1);
4627 
4628 		/* Get the latest version of the reader write value */
4629 		smp_rmb();
4630 	}
4631 
4632 	/* The writer is not moving forward? Something is wrong */
4633 	if (RB_WARN_ON(cpu_buffer, nr_loops == USECS_WAIT))
4634 		reader = NULL;
4635 
4636 	/*
4637 	 * Make sure we see any padding after the write update
4638 	 * (see rb_reset_tail()).
4639 	 *
4640 	 * In addition, a writer may be writing on the reader page
4641 	 * if the page has not been fully filled, so the read barrier
4642 	 * is also needed to make sure we see the content of what is
4643 	 * committed by the writer (see rb_set_commit_to_write()).
4644 	 */
4645 	smp_rmb();
4646 
4647 
4648 	return reader;
4649 }
4650 
4651 static void rb_advance_reader(struct ring_buffer_per_cpu *cpu_buffer)
4652 {
4653 	struct ring_buffer_event *event;
4654 	struct buffer_page *reader;
4655 	unsigned length;
4656 
4657 	reader = rb_get_reader_page(cpu_buffer);
4658 
4659 	/* This function should not be called when buffer is empty */
4660 	if (RB_WARN_ON(cpu_buffer, !reader))
4661 		return;
4662 
4663 	event = rb_reader_event(cpu_buffer);
4664 
4665 	if (event->type_len <= RINGBUF_TYPE_DATA_TYPE_LEN_MAX)
4666 		cpu_buffer->read++;
4667 
4668 	rb_update_read_stamp(cpu_buffer, event);
4669 
4670 	length = rb_event_length(event);
4671 	cpu_buffer->reader_page->read += length;
4672 	cpu_buffer->read_bytes += length;
4673 }
4674 
4675 static void rb_advance_iter(struct ring_buffer_iter *iter)
4676 {
4677 	struct ring_buffer_per_cpu *cpu_buffer;
4678 
4679 	cpu_buffer = iter->cpu_buffer;
4680 
4681 	/* If head == next_event then we need to jump to the next event */
4682 	if (iter->head == iter->next_event) {
4683 		/* If the event gets overwritten again, there's nothing to do */
4684 		if (rb_iter_head_event(iter) == NULL)
4685 			return;
4686 	}
4687 
4688 	iter->head = iter->next_event;
4689 
4690 	/*
4691 	 * Check if we are at the end of the buffer.
4692 	 */
4693 	if (iter->next_event >= rb_page_size(iter->head_page)) {
4694 		/* discarded commits can make the page empty */
4695 		if (iter->head_page == cpu_buffer->commit_page)
4696 			return;
4697 		rb_inc_iter(iter);
4698 		return;
4699 	}
4700 
4701 	rb_update_iter_read_stamp(iter, iter->event);
4702 }
4703 
4704 static int rb_lost_events(struct ring_buffer_per_cpu *cpu_buffer)
4705 {
4706 	return cpu_buffer->lost_events;
4707 }
4708 
4709 static struct ring_buffer_event *
4710 rb_buffer_peek(struct ring_buffer_per_cpu *cpu_buffer, u64 *ts,
4711 	       unsigned long *lost_events)
4712 {
4713 	struct ring_buffer_event *event;
4714 	struct buffer_page *reader;
4715 	int nr_loops = 0;
4716 
4717 	if (ts)
4718 		*ts = 0;
4719  again:
4720 	/*
4721 	 * We repeat when a time extend is encountered.
4722 	 * Since the time extend is always attached to a data event,
4723 	 * we should never loop more than once.
4724 	 * (We never hit the following condition more than twice).
4725 	 */
4726 	if (RB_WARN_ON(cpu_buffer, ++nr_loops > 2))
4727 		return NULL;
4728 
4729 	reader = rb_get_reader_page(cpu_buffer);
4730 	if (!reader)
4731 		return NULL;
4732 
4733 	event = rb_reader_event(cpu_buffer);
4734 
4735 	switch (event->type_len) {
4736 	case RINGBUF_TYPE_PADDING:
4737 		if (rb_null_event(event))
4738 			RB_WARN_ON(cpu_buffer, 1);
4739 		/*
4740 		 * Because the writer could be discarding every
4741 		 * event it creates (which would probably be bad)
4742 		 * if we were to go back to "again" then we may never
4743 		 * catch up, and will trigger the warn on, or lock
4744 		 * the box. Return the padding, and we will release
4745 		 * the current locks, and try again.
4746 		 */
4747 		return event;
4748 
4749 	case RINGBUF_TYPE_TIME_EXTEND:
4750 		/* Internal data, OK to advance */
4751 		rb_advance_reader(cpu_buffer);
4752 		goto again;
4753 
4754 	case RINGBUF_TYPE_TIME_STAMP:
4755 		if (ts) {
4756 			*ts = rb_event_time_stamp(event);
4757 			*ts = rb_fix_abs_ts(*ts, reader->page->time_stamp);
4758 			ring_buffer_normalize_time_stamp(cpu_buffer->buffer,
4759 							 cpu_buffer->cpu, ts);
4760 		}
4761 		/* Internal data, OK to advance */
4762 		rb_advance_reader(cpu_buffer);
4763 		goto again;
4764 
4765 	case RINGBUF_TYPE_DATA:
4766 		if (ts && !(*ts)) {
4767 			*ts = cpu_buffer->read_stamp + event->time_delta;
4768 			ring_buffer_normalize_time_stamp(cpu_buffer->buffer,
4769 							 cpu_buffer->cpu, ts);
4770 		}
4771 		if (lost_events)
4772 			*lost_events = rb_lost_events(cpu_buffer);
4773 		return event;
4774 
4775 	default:
4776 		RB_WARN_ON(cpu_buffer, 1);
4777 	}
4778 
4779 	return NULL;
4780 }
4781 EXPORT_SYMBOL_GPL(ring_buffer_peek);
4782 
4783 static struct ring_buffer_event *
4784 rb_iter_peek(struct ring_buffer_iter *iter, u64 *ts)
4785 {
4786 	struct trace_buffer *buffer;
4787 	struct ring_buffer_per_cpu *cpu_buffer;
4788 	struct ring_buffer_event *event;
4789 	int nr_loops = 0;
4790 
4791 	if (ts)
4792 		*ts = 0;
4793 
4794 	cpu_buffer = iter->cpu_buffer;
4795 	buffer = cpu_buffer->buffer;
4796 
4797 	/*
4798 	 * Check if someone performed a consuming read to the buffer
4799 	 * or removed some pages from the buffer. In these cases,
4800 	 * iterator was invalidated and we need to reset it.
4801 	 */
4802 	if (unlikely(iter->cache_read != cpu_buffer->read ||
4803 		     iter->cache_reader_page != cpu_buffer->reader_page ||
4804 		     iter->cache_pages_removed != cpu_buffer->pages_removed))
4805 		rb_iter_reset(iter);
4806 
4807  again:
4808 	if (ring_buffer_iter_empty(iter))
4809 		return NULL;
4810 
4811 	/*
4812 	 * As the writer can mess with what the iterator is trying
4813 	 * to read, just give up if we fail to get an event after
4814 	 * three tries. The iterator is not as reliable when reading
4815 	 * the ring buffer with an active write as the consumer is.
4816 	 * Do not warn if the three failures is reached.
4817 	 */
4818 	if (++nr_loops > 3)
4819 		return NULL;
4820 
4821 	if (rb_per_cpu_empty(cpu_buffer))
4822 		return NULL;
4823 
4824 	if (iter->head >= rb_page_size(iter->head_page)) {
4825 		rb_inc_iter(iter);
4826 		goto again;
4827 	}
4828 
4829 	event = rb_iter_head_event(iter);
4830 	if (!event)
4831 		goto again;
4832 
4833 	switch (event->type_len) {
4834 	case RINGBUF_TYPE_PADDING:
4835 		if (rb_null_event(event)) {
4836 			rb_inc_iter(iter);
4837 			goto again;
4838 		}
4839 		rb_advance_iter(iter);
4840 		return event;
4841 
4842 	case RINGBUF_TYPE_TIME_EXTEND:
4843 		/* Internal data, OK to advance */
4844 		rb_advance_iter(iter);
4845 		goto again;
4846 
4847 	case RINGBUF_TYPE_TIME_STAMP:
4848 		if (ts) {
4849 			*ts = rb_event_time_stamp(event);
4850 			*ts = rb_fix_abs_ts(*ts, iter->head_page->page->time_stamp);
4851 			ring_buffer_normalize_time_stamp(cpu_buffer->buffer,
4852 							 cpu_buffer->cpu, ts);
4853 		}
4854 		/* Internal data, OK to advance */
4855 		rb_advance_iter(iter);
4856 		goto again;
4857 
4858 	case RINGBUF_TYPE_DATA:
4859 		if (ts && !(*ts)) {
4860 			*ts = iter->read_stamp + event->time_delta;
4861 			ring_buffer_normalize_time_stamp(buffer,
4862 							 cpu_buffer->cpu, ts);
4863 		}
4864 		return event;
4865 
4866 	default:
4867 		RB_WARN_ON(cpu_buffer, 1);
4868 	}
4869 
4870 	return NULL;
4871 }
4872 EXPORT_SYMBOL_GPL(ring_buffer_iter_peek);
4873 
4874 static inline bool rb_reader_lock(struct ring_buffer_per_cpu *cpu_buffer)
4875 {
4876 	if (likely(!in_nmi())) {
4877 		raw_spin_lock(&cpu_buffer->reader_lock);
4878 		return true;
4879 	}
4880 
4881 	/*
4882 	 * If an NMI die dumps out the content of the ring buffer
4883 	 * trylock must be used to prevent a deadlock if the NMI
4884 	 * preempted a task that holds the ring buffer locks. If
4885 	 * we get the lock then all is fine, if not, then continue
4886 	 * to do the read, but this can corrupt the ring buffer,
4887 	 * so it must be permanently disabled from future writes.
4888 	 * Reading from NMI is a oneshot deal.
4889 	 */
4890 	if (raw_spin_trylock(&cpu_buffer->reader_lock))
4891 		return true;
4892 
4893 	/* Continue without locking, but disable the ring buffer */
4894 	atomic_inc(&cpu_buffer->record_disabled);
4895 	return false;
4896 }
4897 
4898 static inline void
4899 rb_reader_unlock(struct ring_buffer_per_cpu *cpu_buffer, bool locked)
4900 {
4901 	if (likely(locked))
4902 		raw_spin_unlock(&cpu_buffer->reader_lock);
4903 }
4904 
4905 /**
4906  * ring_buffer_peek - peek at the next event to be read
4907  * @buffer: The ring buffer to read
4908  * @cpu: The cpu to peak at
4909  * @ts: The timestamp counter of this event.
4910  * @lost_events: a variable to store if events were lost (may be NULL)
4911  *
4912  * This will return the event that will be read next, but does
4913  * not consume the data.
4914  */
4915 struct ring_buffer_event *
4916 ring_buffer_peek(struct trace_buffer *buffer, int cpu, u64 *ts,
4917 		 unsigned long *lost_events)
4918 {
4919 	struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
4920 	struct ring_buffer_event *event;
4921 	unsigned long flags;
4922 	bool dolock;
4923 
4924 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
4925 		return NULL;
4926 
4927  again:
4928 	local_irq_save(flags);
4929 	dolock = rb_reader_lock(cpu_buffer);
4930 	event = rb_buffer_peek(cpu_buffer, ts, lost_events);
4931 	if (event && event->type_len == RINGBUF_TYPE_PADDING)
4932 		rb_advance_reader(cpu_buffer);
4933 	rb_reader_unlock(cpu_buffer, dolock);
4934 	local_irq_restore(flags);
4935 
4936 	if (event && event->type_len == RINGBUF_TYPE_PADDING)
4937 		goto again;
4938 
4939 	return event;
4940 }
4941 
4942 /** ring_buffer_iter_dropped - report if there are dropped events
4943  * @iter: The ring buffer iterator
4944  *
4945  * Returns true if there was dropped events since the last peek.
4946  */
4947 bool ring_buffer_iter_dropped(struct ring_buffer_iter *iter)
4948 {
4949 	bool ret = iter->missed_events != 0;
4950 
4951 	iter->missed_events = 0;
4952 	return ret;
4953 }
4954 EXPORT_SYMBOL_GPL(ring_buffer_iter_dropped);
4955 
4956 /**
4957  * ring_buffer_iter_peek - peek at the next event to be read
4958  * @iter: The ring buffer iterator
4959  * @ts: The timestamp counter of this event.
4960  *
4961  * This will return the event that will be read next, but does
4962  * not increment the iterator.
4963  */
4964 struct ring_buffer_event *
4965 ring_buffer_iter_peek(struct ring_buffer_iter *iter, u64 *ts)
4966 {
4967 	struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
4968 	struct ring_buffer_event *event;
4969 	unsigned long flags;
4970 
4971  again:
4972 	raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
4973 	event = rb_iter_peek(iter, ts);
4974 	raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
4975 
4976 	if (event && event->type_len == RINGBUF_TYPE_PADDING)
4977 		goto again;
4978 
4979 	return event;
4980 }
4981 
4982 /**
4983  * ring_buffer_consume - return an event and consume it
4984  * @buffer: The ring buffer to get the next event from
4985  * @cpu: the cpu to read the buffer from
4986  * @ts: a variable to store the timestamp (may be NULL)
4987  * @lost_events: a variable to store if events were lost (may be NULL)
4988  *
4989  * Returns the next event in the ring buffer, and that event is consumed.
4990  * Meaning, that sequential reads will keep returning a different event,
4991  * and eventually empty the ring buffer if the producer is slower.
4992  */
4993 struct ring_buffer_event *
4994 ring_buffer_consume(struct trace_buffer *buffer, int cpu, u64 *ts,
4995 		    unsigned long *lost_events)
4996 {
4997 	struct ring_buffer_per_cpu *cpu_buffer;
4998 	struct ring_buffer_event *event = NULL;
4999 	unsigned long flags;
5000 	bool dolock;
5001 
5002  again:
5003 	/* might be called in atomic */
5004 	preempt_disable();
5005 
5006 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
5007 		goto out;
5008 
5009 	cpu_buffer = buffer->buffers[cpu];
5010 	local_irq_save(flags);
5011 	dolock = rb_reader_lock(cpu_buffer);
5012 
5013 	event = rb_buffer_peek(cpu_buffer, ts, lost_events);
5014 	if (event) {
5015 		cpu_buffer->lost_events = 0;
5016 		rb_advance_reader(cpu_buffer);
5017 	}
5018 
5019 	rb_reader_unlock(cpu_buffer, dolock);
5020 	local_irq_restore(flags);
5021 
5022  out:
5023 	preempt_enable();
5024 
5025 	if (event && event->type_len == RINGBUF_TYPE_PADDING)
5026 		goto again;
5027 
5028 	return event;
5029 }
5030 EXPORT_SYMBOL_GPL(ring_buffer_consume);
5031 
5032 /**
5033  * ring_buffer_read_prepare - Prepare for a non consuming read of the buffer
5034  * @buffer: The ring buffer to read from
5035  * @cpu: The cpu buffer to iterate over
5036  * @flags: gfp flags to use for memory allocation
5037  *
5038  * This performs the initial preparations necessary to iterate
5039  * through the buffer.  Memory is allocated, buffer recording
5040  * is disabled, and the iterator pointer is returned to the caller.
5041  *
5042  * Disabling buffer recording prevents the reading from being
5043  * corrupted. This is not a consuming read, so a producer is not
5044  * expected.
5045  *
5046  * After a sequence of ring_buffer_read_prepare calls, the user is
5047  * expected to make at least one call to ring_buffer_read_prepare_sync.
5048  * Afterwards, ring_buffer_read_start is invoked to get things going
5049  * for real.
5050  *
5051  * This overall must be paired with ring_buffer_read_finish.
5052  */
5053 struct ring_buffer_iter *
5054 ring_buffer_read_prepare(struct trace_buffer *buffer, int cpu, gfp_t flags)
5055 {
5056 	struct ring_buffer_per_cpu *cpu_buffer;
5057 	struct ring_buffer_iter *iter;
5058 
5059 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
5060 		return NULL;
5061 
5062 	iter = kzalloc(sizeof(*iter), flags);
5063 	if (!iter)
5064 		return NULL;
5065 
5066 	/* Holds the entire event: data and meta data */
5067 	iter->event_size = buffer->subbuf_size;
5068 	iter->event = kmalloc(iter->event_size, flags);
5069 	if (!iter->event) {
5070 		kfree(iter);
5071 		return NULL;
5072 	}
5073 
5074 	cpu_buffer = buffer->buffers[cpu];
5075 
5076 	iter->cpu_buffer = cpu_buffer;
5077 
5078 	atomic_inc(&cpu_buffer->resize_disabled);
5079 
5080 	return iter;
5081 }
5082 EXPORT_SYMBOL_GPL(ring_buffer_read_prepare);
5083 
5084 /**
5085  * ring_buffer_read_prepare_sync - Synchronize a set of prepare calls
5086  *
5087  * All previously invoked ring_buffer_read_prepare calls to prepare
5088  * iterators will be synchronized.  Afterwards, read_buffer_read_start
5089  * calls on those iterators are allowed.
5090  */
5091 void
5092 ring_buffer_read_prepare_sync(void)
5093 {
5094 	synchronize_rcu();
5095 }
5096 EXPORT_SYMBOL_GPL(ring_buffer_read_prepare_sync);
5097 
5098 /**
5099  * ring_buffer_read_start - start a non consuming read of the buffer
5100  * @iter: The iterator returned by ring_buffer_read_prepare
5101  *
5102  * This finalizes the startup of an iteration through the buffer.
5103  * The iterator comes from a call to ring_buffer_read_prepare and
5104  * an intervening ring_buffer_read_prepare_sync must have been
5105  * performed.
5106  *
5107  * Must be paired with ring_buffer_read_finish.
5108  */
5109 void
5110 ring_buffer_read_start(struct ring_buffer_iter *iter)
5111 {
5112 	struct ring_buffer_per_cpu *cpu_buffer;
5113 	unsigned long flags;
5114 
5115 	if (!iter)
5116 		return;
5117 
5118 	cpu_buffer = iter->cpu_buffer;
5119 
5120 	raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
5121 	arch_spin_lock(&cpu_buffer->lock);
5122 	rb_iter_reset(iter);
5123 	arch_spin_unlock(&cpu_buffer->lock);
5124 	raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
5125 }
5126 EXPORT_SYMBOL_GPL(ring_buffer_read_start);
5127 
5128 /**
5129  * ring_buffer_read_finish - finish reading the iterator of the buffer
5130  * @iter: The iterator retrieved by ring_buffer_start
5131  *
5132  * This re-enables the recording to the buffer, and frees the
5133  * iterator.
5134  */
5135 void
5136 ring_buffer_read_finish(struct ring_buffer_iter *iter)
5137 {
5138 	struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
5139 	unsigned long flags;
5140 
5141 	/*
5142 	 * Ring buffer is disabled from recording, here's a good place
5143 	 * to check the integrity of the ring buffer.
5144 	 * Must prevent readers from trying to read, as the check
5145 	 * clears the HEAD page and readers require it.
5146 	 */
5147 	raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
5148 	rb_check_pages(cpu_buffer);
5149 	raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
5150 
5151 	atomic_dec(&cpu_buffer->resize_disabled);
5152 	kfree(iter->event);
5153 	kfree(iter);
5154 }
5155 EXPORT_SYMBOL_GPL(ring_buffer_read_finish);
5156 
5157 /**
5158  * ring_buffer_iter_advance - advance the iterator to the next location
5159  * @iter: The ring buffer iterator
5160  *
5161  * Move the location of the iterator such that the next read will
5162  * be the next location of the iterator.
5163  */
5164 void ring_buffer_iter_advance(struct ring_buffer_iter *iter)
5165 {
5166 	struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
5167 	unsigned long flags;
5168 
5169 	raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
5170 
5171 	rb_advance_iter(iter);
5172 
5173 	raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
5174 }
5175 EXPORT_SYMBOL_GPL(ring_buffer_iter_advance);
5176 
5177 /**
5178  * ring_buffer_size - return the size of the ring buffer (in bytes)
5179  * @buffer: The ring buffer.
5180  * @cpu: The CPU to get ring buffer size from.
5181  */
5182 unsigned long ring_buffer_size(struct trace_buffer *buffer, int cpu)
5183 {
5184 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
5185 		return 0;
5186 
5187 	return buffer->subbuf_size * buffer->buffers[cpu]->nr_pages;
5188 }
5189 EXPORT_SYMBOL_GPL(ring_buffer_size);
5190 
5191 /**
5192  * ring_buffer_max_event_size - return the max data size of an event
5193  * @buffer: The ring buffer.
5194  *
5195  * Returns the maximum size an event can be.
5196  */
5197 unsigned long ring_buffer_max_event_size(struct trace_buffer *buffer)
5198 {
5199 	/* If abs timestamp is requested, events have a timestamp too */
5200 	if (ring_buffer_time_stamp_abs(buffer))
5201 		return buffer->max_data_size - RB_LEN_TIME_EXTEND;
5202 	return buffer->max_data_size;
5203 }
5204 EXPORT_SYMBOL_GPL(ring_buffer_max_event_size);
5205 
5206 static void rb_clear_buffer_page(struct buffer_page *page)
5207 {
5208 	local_set(&page->write, 0);
5209 	local_set(&page->entries, 0);
5210 	rb_init_page(page->page);
5211 	page->read = 0;
5212 }
5213 
5214 static void
5215 rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer)
5216 {
5217 	struct buffer_page *page;
5218 
5219 	rb_head_page_deactivate(cpu_buffer);
5220 
5221 	cpu_buffer->head_page
5222 		= list_entry(cpu_buffer->pages, struct buffer_page, list);
5223 	rb_clear_buffer_page(cpu_buffer->head_page);
5224 	list_for_each_entry(page, cpu_buffer->pages, list) {
5225 		rb_clear_buffer_page(page);
5226 	}
5227 
5228 	cpu_buffer->tail_page = cpu_buffer->head_page;
5229 	cpu_buffer->commit_page = cpu_buffer->head_page;
5230 
5231 	INIT_LIST_HEAD(&cpu_buffer->reader_page->list);
5232 	INIT_LIST_HEAD(&cpu_buffer->new_pages);
5233 	rb_clear_buffer_page(cpu_buffer->reader_page);
5234 
5235 	local_set(&cpu_buffer->entries_bytes, 0);
5236 	local_set(&cpu_buffer->overrun, 0);
5237 	local_set(&cpu_buffer->commit_overrun, 0);
5238 	local_set(&cpu_buffer->dropped_events, 0);
5239 	local_set(&cpu_buffer->entries, 0);
5240 	local_set(&cpu_buffer->committing, 0);
5241 	local_set(&cpu_buffer->commits, 0);
5242 	local_set(&cpu_buffer->pages_touched, 0);
5243 	local_set(&cpu_buffer->pages_lost, 0);
5244 	local_set(&cpu_buffer->pages_read, 0);
5245 	cpu_buffer->last_pages_touch = 0;
5246 	cpu_buffer->shortest_full = 0;
5247 	cpu_buffer->read = 0;
5248 	cpu_buffer->read_bytes = 0;
5249 
5250 	rb_time_set(&cpu_buffer->write_stamp, 0);
5251 	rb_time_set(&cpu_buffer->before_stamp, 0);
5252 
5253 	memset(cpu_buffer->event_stamp, 0, sizeof(cpu_buffer->event_stamp));
5254 
5255 	cpu_buffer->lost_events = 0;
5256 	cpu_buffer->last_overrun = 0;
5257 
5258 	rb_head_page_activate(cpu_buffer);
5259 	cpu_buffer->pages_removed = 0;
5260 }
5261 
5262 /* Must have disabled the cpu buffer then done a synchronize_rcu */
5263 static void reset_disabled_cpu_buffer(struct ring_buffer_per_cpu *cpu_buffer)
5264 {
5265 	unsigned long flags;
5266 
5267 	raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
5268 
5269 	if (RB_WARN_ON(cpu_buffer, local_read(&cpu_buffer->committing)))
5270 		goto out;
5271 
5272 	arch_spin_lock(&cpu_buffer->lock);
5273 
5274 	rb_reset_cpu(cpu_buffer);
5275 
5276 	arch_spin_unlock(&cpu_buffer->lock);
5277 
5278  out:
5279 	raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
5280 }
5281 
5282 /**
5283  * ring_buffer_reset_cpu - reset a ring buffer per CPU buffer
5284  * @buffer: The ring buffer to reset a per cpu buffer of
5285  * @cpu: The CPU buffer to be reset
5286  */
5287 void ring_buffer_reset_cpu(struct trace_buffer *buffer, int cpu)
5288 {
5289 	struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
5290 
5291 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
5292 		return;
5293 
5294 	/* prevent another thread from changing buffer sizes */
5295 	mutex_lock(&buffer->mutex);
5296 
5297 	atomic_inc(&cpu_buffer->resize_disabled);
5298 	atomic_inc(&cpu_buffer->record_disabled);
5299 
5300 	/* Make sure all commits have finished */
5301 	synchronize_rcu();
5302 
5303 	reset_disabled_cpu_buffer(cpu_buffer);
5304 
5305 	atomic_dec(&cpu_buffer->record_disabled);
5306 	atomic_dec(&cpu_buffer->resize_disabled);
5307 
5308 	mutex_unlock(&buffer->mutex);
5309 }
5310 EXPORT_SYMBOL_GPL(ring_buffer_reset_cpu);
5311 
5312 /* Flag to ensure proper resetting of atomic variables */
5313 #define RESET_BIT	(1 << 30)
5314 
5315 /**
5316  * ring_buffer_reset_online_cpus - reset a ring buffer per CPU buffer
5317  * @buffer: The ring buffer to reset a per cpu buffer of
5318  */
5319 void ring_buffer_reset_online_cpus(struct trace_buffer *buffer)
5320 {
5321 	struct ring_buffer_per_cpu *cpu_buffer;
5322 	int cpu;
5323 
5324 	/* prevent another thread from changing buffer sizes */
5325 	mutex_lock(&buffer->mutex);
5326 
5327 	for_each_online_buffer_cpu(buffer, cpu) {
5328 		cpu_buffer = buffer->buffers[cpu];
5329 
5330 		atomic_add(RESET_BIT, &cpu_buffer->resize_disabled);
5331 		atomic_inc(&cpu_buffer->record_disabled);
5332 	}
5333 
5334 	/* Make sure all commits have finished */
5335 	synchronize_rcu();
5336 
5337 	for_each_buffer_cpu(buffer, cpu) {
5338 		cpu_buffer = buffer->buffers[cpu];
5339 
5340 		/*
5341 		 * If a CPU came online during the synchronize_rcu(), then
5342 		 * ignore it.
5343 		 */
5344 		if (!(atomic_read(&cpu_buffer->resize_disabled) & RESET_BIT))
5345 			continue;
5346 
5347 		reset_disabled_cpu_buffer(cpu_buffer);
5348 
5349 		atomic_dec(&cpu_buffer->record_disabled);
5350 		atomic_sub(RESET_BIT, &cpu_buffer->resize_disabled);
5351 	}
5352 
5353 	mutex_unlock(&buffer->mutex);
5354 }
5355 
5356 /**
5357  * ring_buffer_reset - reset a ring buffer
5358  * @buffer: The ring buffer to reset all cpu buffers
5359  */
5360 void ring_buffer_reset(struct trace_buffer *buffer)
5361 {
5362 	struct ring_buffer_per_cpu *cpu_buffer;
5363 	int cpu;
5364 
5365 	/* prevent another thread from changing buffer sizes */
5366 	mutex_lock(&buffer->mutex);
5367 
5368 	for_each_buffer_cpu(buffer, cpu) {
5369 		cpu_buffer = buffer->buffers[cpu];
5370 
5371 		atomic_inc(&cpu_buffer->resize_disabled);
5372 		atomic_inc(&cpu_buffer->record_disabled);
5373 	}
5374 
5375 	/* Make sure all commits have finished */
5376 	synchronize_rcu();
5377 
5378 	for_each_buffer_cpu(buffer, cpu) {
5379 		cpu_buffer = buffer->buffers[cpu];
5380 
5381 		reset_disabled_cpu_buffer(cpu_buffer);
5382 
5383 		atomic_dec(&cpu_buffer->record_disabled);
5384 		atomic_dec(&cpu_buffer->resize_disabled);
5385 	}
5386 
5387 	mutex_unlock(&buffer->mutex);
5388 }
5389 EXPORT_SYMBOL_GPL(ring_buffer_reset);
5390 
5391 /**
5392  * ring_buffer_empty - is the ring buffer empty?
5393  * @buffer: The ring buffer to test
5394  */
5395 bool ring_buffer_empty(struct trace_buffer *buffer)
5396 {
5397 	struct ring_buffer_per_cpu *cpu_buffer;
5398 	unsigned long flags;
5399 	bool dolock;
5400 	bool ret;
5401 	int cpu;
5402 
5403 	/* yes this is racy, but if you don't like the race, lock the buffer */
5404 	for_each_buffer_cpu(buffer, cpu) {
5405 		cpu_buffer = buffer->buffers[cpu];
5406 		local_irq_save(flags);
5407 		dolock = rb_reader_lock(cpu_buffer);
5408 		ret = rb_per_cpu_empty(cpu_buffer);
5409 		rb_reader_unlock(cpu_buffer, dolock);
5410 		local_irq_restore(flags);
5411 
5412 		if (!ret)
5413 			return false;
5414 	}
5415 
5416 	return true;
5417 }
5418 EXPORT_SYMBOL_GPL(ring_buffer_empty);
5419 
5420 /**
5421  * ring_buffer_empty_cpu - is a cpu buffer of a ring buffer empty?
5422  * @buffer: The ring buffer
5423  * @cpu: The CPU buffer to test
5424  */
5425 bool ring_buffer_empty_cpu(struct trace_buffer *buffer, int cpu)
5426 {
5427 	struct ring_buffer_per_cpu *cpu_buffer;
5428 	unsigned long flags;
5429 	bool dolock;
5430 	bool ret;
5431 
5432 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
5433 		return true;
5434 
5435 	cpu_buffer = buffer->buffers[cpu];
5436 	local_irq_save(flags);
5437 	dolock = rb_reader_lock(cpu_buffer);
5438 	ret = rb_per_cpu_empty(cpu_buffer);
5439 	rb_reader_unlock(cpu_buffer, dolock);
5440 	local_irq_restore(flags);
5441 
5442 	return ret;
5443 }
5444 EXPORT_SYMBOL_GPL(ring_buffer_empty_cpu);
5445 
5446 #ifdef CONFIG_RING_BUFFER_ALLOW_SWAP
5447 /**
5448  * ring_buffer_swap_cpu - swap a CPU buffer between two ring buffers
5449  * @buffer_a: One buffer to swap with
5450  * @buffer_b: The other buffer to swap with
5451  * @cpu: the CPU of the buffers to swap
5452  *
5453  * This function is useful for tracers that want to take a "snapshot"
5454  * of a CPU buffer and has another back up buffer lying around.
5455  * it is expected that the tracer handles the cpu buffer not being
5456  * used at the moment.
5457  */
5458 int ring_buffer_swap_cpu(struct trace_buffer *buffer_a,
5459 			 struct trace_buffer *buffer_b, int cpu)
5460 {
5461 	struct ring_buffer_per_cpu *cpu_buffer_a;
5462 	struct ring_buffer_per_cpu *cpu_buffer_b;
5463 	int ret = -EINVAL;
5464 
5465 	if (!cpumask_test_cpu(cpu, buffer_a->cpumask) ||
5466 	    !cpumask_test_cpu(cpu, buffer_b->cpumask))
5467 		goto out;
5468 
5469 	cpu_buffer_a = buffer_a->buffers[cpu];
5470 	cpu_buffer_b = buffer_b->buffers[cpu];
5471 
5472 	/* At least make sure the two buffers are somewhat the same */
5473 	if (cpu_buffer_a->nr_pages != cpu_buffer_b->nr_pages)
5474 		goto out;
5475 
5476 	if (buffer_a->subbuf_order != buffer_b->subbuf_order)
5477 		goto out;
5478 
5479 	ret = -EAGAIN;
5480 
5481 	if (atomic_read(&buffer_a->record_disabled))
5482 		goto out;
5483 
5484 	if (atomic_read(&buffer_b->record_disabled))
5485 		goto out;
5486 
5487 	if (atomic_read(&cpu_buffer_a->record_disabled))
5488 		goto out;
5489 
5490 	if (atomic_read(&cpu_buffer_b->record_disabled))
5491 		goto out;
5492 
5493 	/*
5494 	 * We can't do a synchronize_rcu here because this
5495 	 * function can be called in atomic context.
5496 	 * Normally this will be called from the same CPU as cpu.
5497 	 * If not it's up to the caller to protect this.
5498 	 */
5499 	atomic_inc(&cpu_buffer_a->record_disabled);
5500 	atomic_inc(&cpu_buffer_b->record_disabled);
5501 
5502 	ret = -EBUSY;
5503 	if (local_read(&cpu_buffer_a->committing))
5504 		goto out_dec;
5505 	if (local_read(&cpu_buffer_b->committing))
5506 		goto out_dec;
5507 
5508 	/*
5509 	 * When resize is in progress, we cannot swap it because
5510 	 * it will mess the state of the cpu buffer.
5511 	 */
5512 	if (atomic_read(&buffer_a->resizing))
5513 		goto out_dec;
5514 	if (atomic_read(&buffer_b->resizing))
5515 		goto out_dec;
5516 
5517 	buffer_a->buffers[cpu] = cpu_buffer_b;
5518 	buffer_b->buffers[cpu] = cpu_buffer_a;
5519 
5520 	cpu_buffer_b->buffer = buffer_a;
5521 	cpu_buffer_a->buffer = buffer_b;
5522 
5523 	ret = 0;
5524 
5525 out_dec:
5526 	atomic_dec(&cpu_buffer_a->record_disabled);
5527 	atomic_dec(&cpu_buffer_b->record_disabled);
5528 out:
5529 	return ret;
5530 }
5531 EXPORT_SYMBOL_GPL(ring_buffer_swap_cpu);
5532 #endif /* CONFIG_RING_BUFFER_ALLOW_SWAP */
5533 
5534 /**
5535  * ring_buffer_alloc_read_page - allocate a page to read from buffer
5536  * @buffer: the buffer to allocate for.
5537  * @cpu: the cpu buffer to allocate.
5538  *
5539  * This function is used in conjunction with ring_buffer_read_page.
5540  * When reading a full page from the ring buffer, these functions
5541  * can be used to speed up the process. The calling function should
5542  * allocate a few pages first with this function. Then when it
5543  * needs to get pages from the ring buffer, it passes the result
5544  * of this function into ring_buffer_read_page, which will swap
5545  * the page that was allocated, with the read page of the buffer.
5546  *
5547  * Returns:
5548  *  The page allocated, or ERR_PTR
5549  */
5550 struct buffer_data_read_page *
5551 ring_buffer_alloc_read_page(struct trace_buffer *buffer, int cpu)
5552 {
5553 	struct ring_buffer_per_cpu *cpu_buffer;
5554 	struct buffer_data_read_page *bpage = NULL;
5555 	unsigned long flags;
5556 	struct page *page;
5557 
5558 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
5559 		return ERR_PTR(-ENODEV);
5560 
5561 	bpage = kzalloc(sizeof(*bpage), GFP_KERNEL);
5562 	if (!bpage)
5563 		return ERR_PTR(-ENOMEM);
5564 
5565 	bpage->order = buffer->subbuf_order;
5566 	cpu_buffer = buffer->buffers[cpu];
5567 	local_irq_save(flags);
5568 	arch_spin_lock(&cpu_buffer->lock);
5569 
5570 	if (cpu_buffer->free_page) {
5571 		bpage->data = cpu_buffer->free_page;
5572 		cpu_buffer->free_page = NULL;
5573 	}
5574 
5575 	arch_spin_unlock(&cpu_buffer->lock);
5576 	local_irq_restore(flags);
5577 
5578 	if (bpage->data)
5579 		goto out;
5580 
5581 	page = alloc_pages_node(cpu_to_node(cpu),
5582 				GFP_KERNEL | __GFP_NORETRY | __GFP_ZERO,
5583 				cpu_buffer->buffer->subbuf_order);
5584 	if (!page) {
5585 		kfree(bpage);
5586 		return ERR_PTR(-ENOMEM);
5587 	}
5588 
5589 	bpage->data = page_address(page);
5590 
5591  out:
5592 	rb_init_page(bpage->data);
5593 
5594 	return bpage;
5595 }
5596 EXPORT_SYMBOL_GPL(ring_buffer_alloc_read_page);
5597 
5598 /**
5599  * ring_buffer_free_read_page - free an allocated read page
5600  * @buffer: the buffer the page was allocate for
5601  * @cpu: the cpu buffer the page came from
5602  * @data_page: the page to free
5603  *
5604  * Free a page allocated from ring_buffer_alloc_read_page.
5605  */
5606 void ring_buffer_free_read_page(struct trace_buffer *buffer, int cpu,
5607 				struct buffer_data_read_page *data_page)
5608 {
5609 	struct ring_buffer_per_cpu *cpu_buffer;
5610 	struct buffer_data_page *bpage = data_page->data;
5611 	struct page *page = virt_to_page(bpage);
5612 	unsigned long flags;
5613 
5614 	if (!buffer || !buffer->buffers || !buffer->buffers[cpu])
5615 		return;
5616 
5617 	cpu_buffer = buffer->buffers[cpu];
5618 
5619 	/*
5620 	 * If the page is still in use someplace else, or order of the page
5621 	 * is different from the subbuffer order of the buffer -
5622 	 * we can't reuse it
5623 	 */
5624 	if (page_ref_count(page) > 1 || data_page->order != buffer->subbuf_order)
5625 		goto out;
5626 
5627 	local_irq_save(flags);
5628 	arch_spin_lock(&cpu_buffer->lock);
5629 
5630 	if (!cpu_buffer->free_page) {
5631 		cpu_buffer->free_page = bpage;
5632 		bpage = NULL;
5633 	}
5634 
5635 	arch_spin_unlock(&cpu_buffer->lock);
5636 	local_irq_restore(flags);
5637 
5638  out:
5639 	free_pages((unsigned long)bpage, data_page->order);
5640 	kfree(data_page);
5641 }
5642 EXPORT_SYMBOL_GPL(ring_buffer_free_read_page);
5643 
5644 /**
5645  * ring_buffer_read_page - extract a page from the ring buffer
5646  * @buffer: buffer to extract from
5647  * @data_page: the page to use allocated from ring_buffer_alloc_read_page
5648  * @len: amount to extract
5649  * @cpu: the cpu of the buffer to extract
5650  * @full: should the extraction only happen when the page is full.
5651  *
5652  * This function will pull out a page from the ring buffer and consume it.
5653  * @data_page must be the address of the variable that was returned
5654  * from ring_buffer_alloc_read_page. This is because the page might be used
5655  * to swap with a page in the ring buffer.
5656  *
5657  * for example:
5658  *	rpage = ring_buffer_alloc_read_page(buffer, cpu);
5659  *	if (IS_ERR(rpage))
5660  *		return PTR_ERR(rpage);
5661  *	ret = ring_buffer_read_page(buffer, rpage, len, cpu, 0);
5662  *	if (ret >= 0)
5663  *		process_page(ring_buffer_read_page_data(rpage), ret);
5664  *	ring_buffer_free_read_page(buffer, cpu, rpage);
5665  *
5666  * When @full is set, the function will not return true unless
5667  * the writer is off the reader page.
5668  *
5669  * Note: it is up to the calling functions to handle sleeps and wakeups.
5670  *  The ring buffer can be used anywhere in the kernel and can not
5671  *  blindly call wake_up. The layer that uses the ring buffer must be
5672  *  responsible for that.
5673  *
5674  * Returns:
5675  *  >=0 if data has been transferred, returns the offset of consumed data.
5676  *  <0 if no data has been transferred.
5677  */
5678 int ring_buffer_read_page(struct trace_buffer *buffer,
5679 			  struct buffer_data_read_page *data_page,
5680 			  size_t len, int cpu, int full)
5681 {
5682 	struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
5683 	struct ring_buffer_event *event;
5684 	struct buffer_data_page *bpage;
5685 	struct buffer_page *reader;
5686 	unsigned long missed_events;
5687 	unsigned long flags;
5688 	unsigned int commit;
5689 	unsigned int read;
5690 	u64 save_timestamp;
5691 	int ret = -1;
5692 
5693 	if (!cpumask_test_cpu(cpu, buffer->cpumask))
5694 		goto out;
5695 
5696 	/*
5697 	 * If len is not big enough to hold the page header, then
5698 	 * we can not copy anything.
5699 	 */
5700 	if (len <= BUF_PAGE_HDR_SIZE)
5701 		goto out;
5702 
5703 	len -= BUF_PAGE_HDR_SIZE;
5704 
5705 	if (!data_page || !data_page->data)
5706 		goto out;
5707 	if (data_page->order != buffer->subbuf_order)
5708 		goto out;
5709 
5710 	bpage = data_page->data;
5711 	if (!bpage)
5712 		goto out;
5713 
5714 	raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
5715 
5716 	reader = rb_get_reader_page(cpu_buffer);
5717 	if (!reader)
5718 		goto out_unlock;
5719 
5720 	event = rb_reader_event(cpu_buffer);
5721 
5722 	read = reader->read;
5723 	commit = rb_page_commit(reader);
5724 
5725 	/* Check if any events were dropped */
5726 	missed_events = cpu_buffer->lost_events;
5727 
5728 	/*
5729 	 * If this page has been partially read or
5730 	 * if len is not big enough to read the rest of the page or
5731 	 * a writer is still on the page, then
5732 	 * we must copy the data from the page to the buffer.
5733 	 * Otherwise, we can simply swap the page with the one passed in.
5734 	 */
5735 	if (read || (len < (commit - read)) ||
5736 	    cpu_buffer->reader_page == cpu_buffer->commit_page) {
5737 		struct buffer_data_page *rpage = cpu_buffer->reader_page->page;
5738 		unsigned int rpos = read;
5739 		unsigned int pos = 0;
5740 		unsigned int size;
5741 
5742 		/*
5743 		 * If a full page is expected, this can still be returned
5744 		 * if there's been a previous partial read and the
5745 		 * rest of the page can be read and the commit page is off
5746 		 * the reader page.
5747 		 */
5748 		if (full &&
5749 		    (!read || (len < (commit - read)) ||
5750 		     cpu_buffer->reader_page == cpu_buffer->commit_page))
5751 			goto out_unlock;
5752 
5753 		if (len > (commit - read))
5754 			len = (commit - read);
5755 
5756 		/* Always keep the time extend and data together */
5757 		size = rb_event_ts_length(event);
5758 
5759 		if (len < size)
5760 			goto out_unlock;
5761 
5762 		/* save the current timestamp, since the user will need it */
5763 		save_timestamp = cpu_buffer->read_stamp;
5764 
5765 		/* Need to copy one event at a time */
5766 		do {
5767 			/* We need the size of one event, because
5768 			 * rb_advance_reader only advances by one event,
5769 			 * whereas rb_event_ts_length may include the size of
5770 			 * one or two events.
5771 			 * We have already ensured there's enough space if this
5772 			 * is a time extend. */
5773 			size = rb_event_length(event);
5774 			memcpy(bpage->data + pos, rpage->data + rpos, size);
5775 
5776 			len -= size;
5777 
5778 			rb_advance_reader(cpu_buffer);
5779 			rpos = reader->read;
5780 			pos += size;
5781 
5782 			if (rpos >= commit)
5783 				break;
5784 
5785 			event = rb_reader_event(cpu_buffer);
5786 			/* Always keep the time extend and data together */
5787 			size = rb_event_ts_length(event);
5788 		} while (len >= size);
5789 
5790 		/* update bpage */
5791 		local_set(&bpage->commit, pos);
5792 		bpage->time_stamp = save_timestamp;
5793 
5794 		/* we copied everything to the beginning */
5795 		read = 0;
5796 	} else {
5797 		/* update the entry counter */
5798 		cpu_buffer->read += rb_page_entries(reader);
5799 		cpu_buffer->read_bytes += rb_page_commit(reader);
5800 
5801 		/* swap the pages */
5802 		rb_init_page(bpage);
5803 		bpage = reader->page;
5804 		reader->page = data_page->data;
5805 		local_set(&reader->write, 0);
5806 		local_set(&reader->entries, 0);
5807 		reader->read = 0;
5808 		data_page->data = bpage;
5809 
5810 		/*
5811 		 * Use the real_end for the data size,
5812 		 * This gives us a chance to store the lost events
5813 		 * on the page.
5814 		 */
5815 		if (reader->real_end)
5816 			local_set(&bpage->commit, reader->real_end);
5817 	}
5818 	ret = read;
5819 
5820 	cpu_buffer->lost_events = 0;
5821 
5822 	commit = local_read(&bpage->commit);
5823 	/*
5824 	 * Set a flag in the commit field if we lost events
5825 	 */
5826 	if (missed_events) {
5827 		/* If there is room at the end of the page to save the
5828 		 * missed events, then record it there.
5829 		 */
5830 		if (buffer->subbuf_size - commit >= sizeof(missed_events)) {
5831 			memcpy(&bpage->data[commit], &missed_events,
5832 			       sizeof(missed_events));
5833 			local_add(RB_MISSED_STORED, &bpage->commit);
5834 			commit += sizeof(missed_events);
5835 		}
5836 		local_add(RB_MISSED_EVENTS, &bpage->commit);
5837 	}
5838 
5839 	/*
5840 	 * This page may be off to user land. Zero it out here.
5841 	 */
5842 	if (commit < buffer->subbuf_size)
5843 		memset(&bpage->data[commit], 0, buffer->subbuf_size - commit);
5844 
5845  out_unlock:
5846 	raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
5847 
5848  out:
5849 	return ret;
5850 }
5851 EXPORT_SYMBOL_GPL(ring_buffer_read_page);
5852 
5853 /**
5854  * ring_buffer_read_page_data - get pointer to the data in the page.
5855  * @page:  the page to get the data from
5856  *
5857  * Returns pointer to the actual data in this page.
5858  */
5859 void *ring_buffer_read_page_data(struct buffer_data_read_page *page)
5860 {
5861 	return page->data;
5862 }
5863 EXPORT_SYMBOL_GPL(ring_buffer_read_page_data);
5864 
5865 /**
5866  * ring_buffer_subbuf_size_get - get size of the sub buffer.
5867  * @buffer: the buffer to get the sub buffer size from
5868  *
5869  * Returns size of the sub buffer, in bytes.
5870  */
5871 int ring_buffer_subbuf_size_get(struct trace_buffer *buffer)
5872 {
5873 	return buffer->subbuf_size + BUF_PAGE_HDR_SIZE;
5874 }
5875 EXPORT_SYMBOL_GPL(ring_buffer_subbuf_size_get);
5876 
5877 /**
5878  * ring_buffer_subbuf_order_get - get order of system sub pages in one buffer page.
5879  * @buffer: The ring_buffer to get the system sub page order from
5880  *
5881  * By default, one ring buffer sub page equals to one system page. This parameter
5882  * is configurable, per ring buffer. The size of the ring buffer sub page can be
5883  * extended, but must be an order of system page size.
5884  *
5885  * Returns the order of buffer sub page size, in system pages:
5886  * 0 means the sub buffer size is 1 system page and so forth.
5887  * In case of an error < 0 is returned.
5888  */
5889 int ring_buffer_subbuf_order_get(struct trace_buffer *buffer)
5890 {
5891 	if (!buffer)
5892 		return -EINVAL;
5893 
5894 	return buffer->subbuf_order;
5895 }
5896 EXPORT_SYMBOL_GPL(ring_buffer_subbuf_order_get);
5897 
5898 /**
5899  * ring_buffer_subbuf_order_set - set the size of ring buffer sub page.
5900  * @buffer: The ring_buffer to set the new page size.
5901  * @order: Order of the system pages in one sub buffer page
5902  *
5903  * By default, one ring buffer pages equals to one system page. This API can be
5904  * used to set new size of the ring buffer page. The size must be order of
5905  * system page size, that's why the input parameter @order is the order of
5906  * system pages that are allocated for one ring buffer page:
5907  *  0 - 1 system page
5908  *  1 - 2 system pages
5909  *  3 - 4 system pages
5910  *  ...
5911  *
5912  * Returns 0 on success or < 0 in case of an error.
5913  */
5914 int ring_buffer_subbuf_order_set(struct trace_buffer *buffer, int order)
5915 {
5916 	struct ring_buffer_per_cpu *cpu_buffer;
5917 	struct buffer_page *bpage, *tmp;
5918 	int old_order, old_size;
5919 	int nr_pages;
5920 	int psize;
5921 	int err;
5922 	int cpu;
5923 
5924 	if (!buffer || order < 0)
5925 		return -EINVAL;
5926 
5927 	if (buffer->subbuf_order == order)
5928 		return 0;
5929 
5930 	psize = (1 << order) * PAGE_SIZE;
5931 	if (psize <= BUF_PAGE_HDR_SIZE)
5932 		return -EINVAL;
5933 
5934 	/* Size of a subbuf cannot be greater than the write counter */
5935 	if (psize > RB_WRITE_MASK + 1)
5936 		return -EINVAL;
5937 
5938 	old_order = buffer->subbuf_order;
5939 	old_size = buffer->subbuf_size;
5940 
5941 	/* prevent another thread from changing buffer sizes */
5942 	mutex_lock(&buffer->mutex);
5943 	atomic_inc(&buffer->record_disabled);
5944 
5945 	/* Make sure all commits have finished */
5946 	synchronize_rcu();
5947 
5948 	buffer->subbuf_order = order;
5949 	buffer->subbuf_size = psize - BUF_PAGE_HDR_SIZE;
5950 
5951 	/* Make sure all new buffers are allocated, before deleting the old ones */
5952 	for_each_buffer_cpu(buffer, cpu) {
5953 
5954 		if (!cpumask_test_cpu(cpu, buffer->cpumask))
5955 			continue;
5956 
5957 		cpu_buffer = buffer->buffers[cpu];
5958 
5959 		/* Update the number of pages to match the new size */
5960 		nr_pages = old_size * buffer->buffers[cpu]->nr_pages;
5961 		nr_pages = DIV_ROUND_UP(nr_pages, buffer->subbuf_size);
5962 
5963 		/* we need a minimum of two pages */
5964 		if (nr_pages < 2)
5965 			nr_pages = 2;
5966 
5967 		cpu_buffer->nr_pages_to_update = nr_pages;
5968 
5969 		/* Include the reader page */
5970 		nr_pages++;
5971 
5972 		/* Allocate the new size buffer */
5973 		INIT_LIST_HEAD(&cpu_buffer->new_pages);
5974 		if (__rb_allocate_pages(cpu_buffer, nr_pages,
5975 					&cpu_buffer->new_pages)) {
5976 			/* not enough memory for new pages */
5977 			err = -ENOMEM;
5978 			goto error;
5979 		}
5980 	}
5981 
5982 	for_each_buffer_cpu(buffer, cpu) {
5983 
5984 		if (!cpumask_test_cpu(cpu, buffer->cpumask))
5985 			continue;
5986 
5987 		cpu_buffer = buffer->buffers[cpu];
5988 
5989 		/* Clear the head bit to make the link list normal to read */
5990 		rb_head_page_deactivate(cpu_buffer);
5991 
5992 		/* Now walk the list and free all the old sub buffers */
5993 		list_for_each_entry_safe(bpage, tmp, cpu_buffer->pages, list) {
5994 			list_del_init(&bpage->list);
5995 			free_buffer_page(bpage);
5996 		}
5997 		/* The above loop stopped an the last page needing to be freed */
5998 		bpage = list_entry(cpu_buffer->pages, struct buffer_page, list);
5999 		free_buffer_page(bpage);
6000 
6001 		/* Free the current reader page */
6002 		free_buffer_page(cpu_buffer->reader_page);
6003 
6004 		/* One page was allocated for the reader page */
6005 		cpu_buffer->reader_page = list_entry(cpu_buffer->new_pages.next,
6006 						     struct buffer_page, list);
6007 		list_del_init(&cpu_buffer->reader_page->list);
6008 
6009 		/* The cpu_buffer pages are a link list with no head */
6010 		cpu_buffer->pages = cpu_buffer->new_pages.next;
6011 		cpu_buffer->new_pages.next->prev = cpu_buffer->new_pages.prev;
6012 		cpu_buffer->new_pages.prev->next = cpu_buffer->new_pages.next;
6013 
6014 		/* Clear the new_pages list */
6015 		INIT_LIST_HEAD(&cpu_buffer->new_pages);
6016 
6017 		cpu_buffer->head_page
6018 			= list_entry(cpu_buffer->pages, struct buffer_page, list);
6019 		cpu_buffer->tail_page = cpu_buffer->commit_page = cpu_buffer->head_page;
6020 
6021 		cpu_buffer->nr_pages = cpu_buffer->nr_pages_to_update;
6022 		cpu_buffer->nr_pages_to_update = 0;
6023 
6024 		free_pages((unsigned long)cpu_buffer->free_page, old_order);
6025 		cpu_buffer->free_page = NULL;
6026 
6027 		rb_head_page_activate(cpu_buffer);
6028 
6029 		rb_check_pages(cpu_buffer);
6030 	}
6031 
6032 	atomic_dec(&buffer->record_disabled);
6033 	mutex_unlock(&buffer->mutex);
6034 
6035 	return 0;
6036 
6037 error:
6038 	buffer->subbuf_order = old_order;
6039 	buffer->subbuf_size = old_size;
6040 
6041 	atomic_dec(&buffer->record_disabled);
6042 	mutex_unlock(&buffer->mutex);
6043 
6044 	for_each_buffer_cpu(buffer, cpu) {
6045 		cpu_buffer = buffer->buffers[cpu];
6046 
6047 		if (!cpu_buffer->nr_pages_to_update)
6048 			continue;
6049 
6050 		list_for_each_entry_safe(bpage, tmp, &cpu_buffer->new_pages, list) {
6051 			list_del_init(&bpage->list);
6052 			free_buffer_page(bpage);
6053 		}
6054 	}
6055 
6056 	return err;
6057 }
6058 EXPORT_SYMBOL_GPL(ring_buffer_subbuf_order_set);
6059 
6060 /*
6061  * We only allocate new buffers, never free them if the CPU goes down.
6062  * If we were to free the buffer, then the user would lose any trace that was in
6063  * the buffer.
6064  */
6065 int trace_rb_cpu_prepare(unsigned int cpu, struct hlist_node *node)
6066 {
6067 	struct trace_buffer *buffer;
6068 	long nr_pages_same;
6069 	int cpu_i;
6070 	unsigned long nr_pages;
6071 
6072 	buffer = container_of(node, struct trace_buffer, node);
6073 	if (cpumask_test_cpu(cpu, buffer->cpumask))
6074 		return 0;
6075 
6076 	nr_pages = 0;
6077 	nr_pages_same = 1;
6078 	/* check if all cpu sizes are same */
6079 	for_each_buffer_cpu(buffer, cpu_i) {
6080 		/* fill in the size from first enabled cpu */
6081 		if (nr_pages == 0)
6082 			nr_pages = buffer->buffers[cpu_i]->nr_pages;
6083 		if (nr_pages != buffer->buffers[cpu_i]->nr_pages) {
6084 			nr_pages_same = 0;
6085 			break;
6086 		}
6087 	}
6088 	/* allocate minimum pages, user can later expand it */
6089 	if (!nr_pages_same)
6090 		nr_pages = 2;
6091 	buffer->buffers[cpu] =
6092 		rb_allocate_cpu_buffer(buffer, nr_pages, cpu);
6093 	if (!buffer->buffers[cpu]) {
6094 		WARN(1, "failed to allocate ring buffer on CPU %u\n",
6095 		     cpu);
6096 		return -ENOMEM;
6097 	}
6098 	smp_wmb();
6099 	cpumask_set_cpu(cpu, buffer->cpumask);
6100 	return 0;
6101 }
6102 
6103 #ifdef CONFIG_RING_BUFFER_STARTUP_TEST
6104 /*
6105  * This is a basic integrity check of the ring buffer.
6106  * Late in the boot cycle this test will run when configured in.
6107  * It will kick off a thread per CPU that will go into a loop
6108  * writing to the per cpu ring buffer various sizes of data.
6109  * Some of the data will be large items, some small.
6110  *
6111  * Another thread is created that goes into a spin, sending out
6112  * IPIs to the other CPUs to also write into the ring buffer.
6113  * this is to test the nesting ability of the buffer.
6114  *
6115  * Basic stats are recorded and reported. If something in the
6116  * ring buffer should happen that's not expected, a big warning
6117  * is displayed and all ring buffers are disabled.
6118  */
6119 static struct task_struct *rb_threads[NR_CPUS] __initdata;
6120 
6121 struct rb_test_data {
6122 	struct trace_buffer *buffer;
6123 	unsigned long		events;
6124 	unsigned long		bytes_written;
6125 	unsigned long		bytes_alloc;
6126 	unsigned long		bytes_dropped;
6127 	unsigned long		events_nested;
6128 	unsigned long		bytes_written_nested;
6129 	unsigned long		bytes_alloc_nested;
6130 	unsigned long		bytes_dropped_nested;
6131 	int			min_size_nested;
6132 	int			max_size_nested;
6133 	int			max_size;
6134 	int			min_size;
6135 	int			cpu;
6136 	int			cnt;
6137 };
6138 
6139 static struct rb_test_data rb_data[NR_CPUS] __initdata;
6140 
6141 /* 1 meg per cpu */
6142 #define RB_TEST_BUFFER_SIZE	1048576
6143 
6144 static char rb_string[] __initdata =
6145 	"abcdefghijklmnopqrstuvwxyz1234567890!@#$%^&*()?+\\"
6146 	"?+|:';\",.<>/?abcdefghijklmnopqrstuvwxyz1234567890"
6147 	"!@#$%^&*()?+\\?+|:';\",.<>/?abcdefghijklmnopqrstuv";
6148 
6149 static bool rb_test_started __initdata;
6150 
6151 struct rb_item {
6152 	int size;
6153 	char str[];
6154 };
6155 
6156 static __init int rb_write_something(struct rb_test_data *data, bool nested)
6157 {
6158 	struct ring_buffer_event *event;
6159 	struct rb_item *item;
6160 	bool started;
6161 	int event_len;
6162 	int size;
6163 	int len;
6164 	int cnt;
6165 
6166 	/* Have nested writes different that what is written */
6167 	cnt = data->cnt + (nested ? 27 : 0);
6168 
6169 	/* Multiply cnt by ~e, to make some unique increment */
6170 	size = (cnt * 68 / 25) % (sizeof(rb_string) - 1);
6171 
6172 	len = size + sizeof(struct rb_item);
6173 
6174 	started = rb_test_started;
6175 	/* read rb_test_started before checking buffer enabled */
6176 	smp_rmb();
6177 
6178 	event = ring_buffer_lock_reserve(data->buffer, len);
6179 	if (!event) {
6180 		/* Ignore dropped events before test starts. */
6181 		if (started) {
6182 			if (nested)
6183 				data->bytes_dropped += len;
6184 			else
6185 				data->bytes_dropped_nested += len;
6186 		}
6187 		return len;
6188 	}
6189 
6190 	event_len = ring_buffer_event_length(event);
6191 
6192 	if (RB_WARN_ON(data->buffer, event_len < len))
6193 		goto out;
6194 
6195 	item = ring_buffer_event_data(event);
6196 	item->size = size;
6197 	memcpy(item->str, rb_string, size);
6198 
6199 	if (nested) {
6200 		data->bytes_alloc_nested += event_len;
6201 		data->bytes_written_nested += len;
6202 		data->events_nested++;
6203 		if (!data->min_size_nested || len < data->min_size_nested)
6204 			data->min_size_nested = len;
6205 		if (len > data->max_size_nested)
6206 			data->max_size_nested = len;
6207 	} else {
6208 		data->bytes_alloc += event_len;
6209 		data->bytes_written += len;
6210 		data->events++;
6211 		if (!data->min_size || len < data->min_size)
6212 			data->max_size = len;
6213 		if (len > data->max_size)
6214 			data->max_size = len;
6215 	}
6216 
6217  out:
6218 	ring_buffer_unlock_commit(data->buffer);
6219 
6220 	return 0;
6221 }
6222 
6223 static __init int rb_test(void *arg)
6224 {
6225 	struct rb_test_data *data = arg;
6226 
6227 	while (!kthread_should_stop()) {
6228 		rb_write_something(data, false);
6229 		data->cnt++;
6230 
6231 		set_current_state(TASK_INTERRUPTIBLE);
6232 		/* Now sleep between a min of 100-300us and a max of 1ms */
6233 		usleep_range(((data->cnt % 3) + 1) * 100, 1000);
6234 	}
6235 
6236 	return 0;
6237 }
6238 
6239 static __init void rb_ipi(void *ignore)
6240 {
6241 	struct rb_test_data *data;
6242 	int cpu = smp_processor_id();
6243 
6244 	data = &rb_data[cpu];
6245 	rb_write_something(data, true);
6246 }
6247 
6248 static __init int rb_hammer_test(void *arg)
6249 {
6250 	while (!kthread_should_stop()) {
6251 
6252 		/* Send an IPI to all cpus to write data! */
6253 		smp_call_function(rb_ipi, NULL, 1);
6254 		/* No sleep, but for non preempt, let others run */
6255 		schedule();
6256 	}
6257 
6258 	return 0;
6259 }
6260 
6261 static __init int test_ringbuffer(void)
6262 {
6263 	struct task_struct *rb_hammer;
6264 	struct trace_buffer *buffer;
6265 	int cpu;
6266 	int ret = 0;
6267 
6268 	if (security_locked_down(LOCKDOWN_TRACEFS)) {
6269 		pr_warn("Lockdown is enabled, skipping ring buffer tests\n");
6270 		return 0;
6271 	}
6272 
6273 	pr_info("Running ring buffer tests...\n");
6274 
6275 	buffer = ring_buffer_alloc(RB_TEST_BUFFER_SIZE, RB_FL_OVERWRITE);
6276 	if (WARN_ON(!buffer))
6277 		return 0;
6278 
6279 	/* Disable buffer so that threads can't write to it yet */
6280 	ring_buffer_record_off(buffer);
6281 
6282 	for_each_online_cpu(cpu) {
6283 		rb_data[cpu].buffer = buffer;
6284 		rb_data[cpu].cpu = cpu;
6285 		rb_data[cpu].cnt = cpu;
6286 		rb_threads[cpu] = kthread_run_on_cpu(rb_test, &rb_data[cpu],
6287 						     cpu, "rbtester/%u");
6288 		if (WARN_ON(IS_ERR(rb_threads[cpu]))) {
6289 			pr_cont("FAILED\n");
6290 			ret = PTR_ERR(rb_threads[cpu]);
6291 			goto out_free;
6292 		}
6293 	}
6294 
6295 	/* Now create the rb hammer! */
6296 	rb_hammer = kthread_run(rb_hammer_test, NULL, "rbhammer");
6297 	if (WARN_ON(IS_ERR(rb_hammer))) {
6298 		pr_cont("FAILED\n");
6299 		ret = PTR_ERR(rb_hammer);
6300 		goto out_free;
6301 	}
6302 
6303 	ring_buffer_record_on(buffer);
6304 	/*
6305 	 * Show buffer is enabled before setting rb_test_started.
6306 	 * Yes there's a small race window where events could be
6307 	 * dropped and the thread wont catch it. But when a ring
6308 	 * buffer gets enabled, there will always be some kind of
6309 	 * delay before other CPUs see it. Thus, we don't care about
6310 	 * those dropped events. We care about events dropped after
6311 	 * the threads see that the buffer is active.
6312 	 */
6313 	smp_wmb();
6314 	rb_test_started = true;
6315 
6316 	set_current_state(TASK_INTERRUPTIBLE);
6317 	/* Just run for 10 seconds */;
6318 	schedule_timeout(10 * HZ);
6319 
6320 	kthread_stop(rb_hammer);
6321 
6322  out_free:
6323 	for_each_online_cpu(cpu) {
6324 		if (!rb_threads[cpu])
6325 			break;
6326 		kthread_stop(rb_threads[cpu]);
6327 	}
6328 	if (ret) {
6329 		ring_buffer_free(buffer);
6330 		return ret;
6331 	}
6332 
6333 	/* Report! */
6334 	pr_info("finished\n");
6335 	for_each_online_cpu(cpu) {
6336 		struct ring_buffer_event *event;
6337 		struct rb_test_data *data = &rb_data[cpu];
6338 		struct rb_item *item;
6339 		unsigned long total_events;
6340 		unsigned long total_dropped;
6341 		unsigned long total_written;
6342 		unsigned long total_alloc;
6343 		unsigned long total_read = 0;
6344 		unsigned long total_size = 0;
6345 		unsigned long total_len = 0;
6346 		unsigned long total_lost = 0;
6347 		unsigned long lost;
6348 		int big_event_size;
6349 		int small_event_size;
6350 
6351 		ret = -1;
6352 
6353 		total_events = data->events + data->events_nested;
6354 		total_written = data->bytes_written + data->bytes_written_nested;
6355 		total_alloc = data->bytes_alloc + data->bytes_alloc_nested;
6356 		total_dropped = data->bytes_dropped + data->bytes_dropped_nested;
6357 
6358 		big_event_size = data->max_size + data->max_size_nested;
6359 		small_event_size = data->min_size + data->min_size_nested;
6360 
6361 		pr_info("CPU %d:\n", cpu);
6362 		pr_info("              events:    %ld\n", total_events);
6363 		pr_info("       dropped bytes:    %ld\n", total_dropped);
6364 		pr_info("       alloced bytes:    %ld\n", total_alloc);
6365 		pr_info("       written bytes:    %ld\n", total_written);
6366 		pr_info("       biggest event:    %d\n", big_event_size);
6367 		pr_info("      smallest event:    %d\n", small_event_size);
6368 
6369 		if (RB_WARN_ON(buffer, total_dropped))
6370 			break;
6371 
6372 		ret = 0;
6373 
6374 		while ((event = ring_buffer_consume(buffer, cpu, NULL, &lost))) {
6375 			total_lost += lost;
6376 			item = ring_buffer_event_data(event);
6377 			total_len += ring_buffer_event_length(event);
6378 			total_size += item->size + sizeof(struct rb_item);
6379 			if (memcmp(&item->str[0], rb_string, item->size) != 0) {
6380 				pr_info("FAILED!\n");
6381 				pr_info("buffer had: %.*s\n", item->size, item->str);
6382 				pr_info("expected:   %.*s\n", item->size, rb_string);
6383 				RB_WARN_ON(buffer, 1);
6384 				ret = -1;
6385 				break;
6386 			}
6387 			total_read++;
6388 		}
6389 		if (ret)
6390 			break;
6391 
6392 		ret = -1;
6393 
6394 		pr_info("         read events:   %ld\n", total_read);
6395 		pr_info("         lost events:   %ld\n", total_lost);
6396 		pr_info("        total events:   %ld\n", total_lost + total_read);
6397 		pr_info("  recorded len bytes:   %ld\n", total_len);
6398 		pr_info(" recorded size bytes:   %ld\n", total_size);
6399 		if (total_lost) {
6400 			pr_info(" With dropped events, record len and size may not match\n"
6401 				" alloced and written from above\n");
6402 		} else {
6403 			if (RB_WARN_ON(buffer, total_len != total_alloc ||
6404 				       total_size != total_written))
6405 				break;
6406 		}
6407 		if (RB_WARN_ON(buffer, total_lost + total_read != total_events))
6408 			break;
6409 
6410 		ret = 0;
6411 	}
6412 	if (!ret)
6413 		pr_info("Ring buffer PASSED!\n");
6414 
6415 	ring_buffer_free(buffer);
6416 	return 0;
6417 }
6418 
6419 late_initcall(test_ringbuffer);
6420 #endif /* CONFIG_RING_BUFFER_STARTUP_TEST */
6421