ring_buffer.c 112.3 KB
Newer Older
S
Steven Rostedt 已提交
1 2 3 4 5 6
/*
 * Generic ring buffer
 *
 * Copyright (C) 2008 Steven Rostedt <srostedt@redhat.com>
 */
#include <linux/ring_buffer.h>
7
#include <linux/trace_clock.h>
S
Steven Rostedt 已提交
8 9 10
#include <linux/spinlock.h>
#include <linux/debugfs.h>
#include <linux/uaccess.h>
11
#include <linux/hardirq.h>
12
#include <linux/kmemcheck.h>
S
Steven Rostedt 已提交
13 14 15
#include <linux/module.h>
#include <linux/percpu.h>
#include <linux/mutex.h>
16
#include <linux/slab.h>
S
Steven Rostedt 已提交
17 18 19
#include <linux/init.h>
#include <linux/hash.h>
#include <linux/list.h>
20
#include <linux/cpu.h>
S
Steven Rostedt 已提交
21 22
#include <linux/fs.h>

23
#include <asm/local.h>
24 25
#include "trace.h"

26 27
static void update_pages_handler(struct work_struct *work);

28 29 30 31 32 33 34
/*
 * The ring buffer header is special. We must manually up keep it.
 */
int ring_buffer_print_entry_header(struct trace_seq *s)
{
	int ret;

35 36
	ret = trace_seq_printf(s, "# compressed entry header\n");
	ret = trace_seq_printf(s, "\ttype_len    :    5 bits\n");
37 38 39 40 41 42 43
	ret = trace_seq_printf(s, "\ttime_delta  :   27 bits\n");
	ret = trace_seq_printf(s, "\tarray       :   32 bits\n");
	ret = trace_seq_printf(s, "\n");
	ret = trace_seq_printf(s, "\tpadding     : type == %d\n",
			       RINGBUF_TYPE_PADDING);
	ret = trace_seq_printf(s, "\ttime_extend : type == %d\n",
			       RINGBUF_TYPE_TIME_EXTEND);
44 45
	ret = trace_seq_printf(s, "\tdata max type_len  == %d\n",
			       RINGBUF_TYPE_DATA_TYPE_LEN_MAX);
46 47 48 49

	return ret;
}

50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117
/*
 * The ring buffer is made up of a list of pages. A separate list of pages is
 * allocated for each CPU. A writer may only write to a buffer that is
 * associated with the CPU it is currently executing on.  A reader may read
 * from any per cpu buffer.
 *
 * The reader is special. For each per cpu buffer, the reader has its own
 * reader page. When a reader has read the entire reader page, this reader
 * page is swapped with another page in the ring buffer.
 *
 * Now, as long as the writer is off the reader page, the reader can do what
 * ever it wants with that page. The writer will never write to that page
 * again (as long as it is out of the ring buffer).
 *
 * Here's some silly ASCII art.
 *
 *   +------+
 *   |reader|          RING BUFFER
 *   |page  |
 *   +------+        +---+   +---+   +---+
 *                   |   |-->|   |-->|   |
 *                   +---+   +---+   +---+
 *                     ^               |
 *                     |               |
 *                     +---------------+
 *
 *
 *   +------+
 *   |reader|          RING BUFFER
 *   |page  |------------------v
 *   +------+        +---+   +---+   +---+
 *                   |   |-->|   |-->|   |
 *                   +---+   +---+   +---+
 *                     ^               |
 *                     |               |
 *                     +---------------+
 *
 *
 *   +------+
 *   |reader|          RING BUFFER
 *   |page  |------------------v
 *   +------+        +---+   +---+   +---+
 *      ^            |   |-->|   |-->|   |
 *      |            +---+   +---+   +---+
 *      |                              |
 *      |                              |
 *      +------------------------------+
 *
 *
 *   +------+
 *   |buffer|          RING BUFFER
 *   |page  |------------------v
 *   +------+        +---+   +---+   +---+
 *      ^            |   |   |   |-->|   |
 *      |   New      +---+   +---+   +---+
 *      |  Reader------^               |
 *      |   page                       |
 *      +------------------------------+
 *
 *
 * After we make this swap, the reader can hand this page off to the splice
 * code and be done with it. It can even allocate a new page if it needs to
 * and swap that into the ring buffer.
 *
 * We will be using cmpxchg soon to make all this lockless.
 *
 */

118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156
/*
 * A fast way to enable or disable all ring buffers is to
 * call tracing_on or tracing_off. Turning off the ring buffers
 * prevents all ring buffers from being recorded to.
 * Turning this switch on, makes it OK to write to the
 * ring buffer, if the ring buffer is enabled itself.
 *
 * There's three layers that must be on in order to write
 * to the ring buffer.
 *
 * 1) This global flag must be set.
 * 2) The ring buffer must be enabled for recording.
 * 3) The per cpu buffer must be enabled for recording.
 *
 * In case of an anomaly, this global flag has a bit set that
 * will permantly disable all ring buffers.
 */

/*
 * Global flag to disable all recording to ring buffers
 *  This has two bits: ON, DISABLED
 *
 *  ON   DISABLED
 * ---- ----------
 *   0      0        : ring buffers are off
 *   1      0        : ring buffers are on
 *   X      1        : ring buffers are permanently disabled
 */

enum {
	RB_BUFFERS_ON_BIT	= 0,
	RB_BUFFERS_DISABLED_BIT	= 1,
};

enum {
	RB_BUFFERS_ON		= 1 << RB_BUFFERS_ON_BIT,
	RB_BUFFERS_DISABLED	= 1 << RB_BUFFERS_DISABLED_BIT,
};

157
static unsigned long ring_buffer_flags __read_mostly = RB_BUFFERS_ON;
158

159 160
/* Used for individual buffers (after the counter) */
#define RB_BUFFER_OFF		(1 << 20)
161

162
#define BUF_PAGE_HDR_SIZE offsetof(struct buffer_data_page, data)
163 164 165 166 167

/**
 * tracing_off_permanent - permanently disable ring buffers
 *
 * This function, once called, will disable all ring buffers
W
Wenji Huang 已提交
168
 * permanently.
169 170 171 172
 */
void tracing_off_permanent(void)
{
	set_bit(RB_BUFFERS_DISABLED_BIT, &ring_buffer_flags);
173 174
}

175
#define RB_EVNT_HDR_SIZE (offsetof(struct ring_buffer_event, array))
176
#define RB_ALIGNMENT		4U
177
#define RB_MAX_SMALL_DATA	(RB_ALIGNMENT * RINGBUF_TYPE_DATA_TYPE_LEN_MAX)
178
#define RB_EVNT_MIN_SIZE	8U	/* two 32bit words */
179

180 181 182 183 184 185 186 187
#if !defined(CONFIG_64BIT) || defined(CONFIG_HAVE_EFFICIENT_UNALIGNED_ACCESS)
# define RB_FORCE_8BYTE_ALIGNMENT	0
# define RB_ARCH_ALIGNMENT		RB_ALIGNMENT
#else
# define RB_FORCE_8BYTE_ALIGNMENT	1
# define RB_ARCH_ALIGNMENT		8U
#endif

188 189
/* define RINGBUF_TYPE_DATA for 'case RINGBUF_TYPE_DATA:' */
#define RINGBUF_TYPE_DATA 0 ... RINGBUF_TYPE_DATA_TYPE_LEN_MAX
S
Steven Rostedt 已提交
190 191 192 193 194 195

enum {
	RB_LEN_TIME_EXTEND = 8,
	RB_LEN_TIME_STAMP = 16,
};

196 197 198
#define skip_time_extend(event) \
	((struct ring_buffer_event *)((char *)event + RB_LEN_TIME_EXTEND))

199 200
static inline int rb_null_event(struct ring_buffer_event *event)
{
201
	return event->type_len == RINGBUF_TYPE_PADDING && !event->time_delta;
202 203 204 205
}

static void rb_event_set_padding(struct ring_buffer_event *event)
{
206
	/* padding has a NULL time_delta */
207
	event->type_len = RINGBUF_TYPE_PADDING;
208 209 210
	event->time_delta = 0;
}

211
static unsigned
212
rb_event_data_length(struct ring_buffer_event *event)
S
Steven Rostedt 已提交
213 214 215
{
	unsigned length;

216 217
	if (event->type_len)
		length = event->type_len * RB_ALIGNMENT;
218 219 220 221 222
	else
		length = event->array[0];
	return length + RB_EVNT_HDR_SIZE;
}

223 224 225 226 227 228
/*
 * Return the length of the given event. Will return
 * the length of the time extend if the event is a
 * time extend.
 */
static inline unsigned
229 230
rb_event_length(struct ring_buffer_event *event)
{
231
	switch (event->type_len) {
S
Steven Rostedt 已提交
232
	case RINGBUF_TYPE_PADDING:
233 234 235
		if (rb_null_event(event))
			/* undefined */
			return -1;
236
		return  event->array[0] + RB_EVNT_HDR_SIZE;
S
Steven Rostedt 已提交
237 238 239 240 241 242 243 244

	case RINGBUF_TYPE_TIME_EXTEND:
		return RB_LEN_TIME_EXTEND;

	case RINGBUF_TYPE_TIME_STAMP:
		return RB_LEN_TIME_STAMP;

	case RINGBUF_TYPE_DATA:
245
		return rb_event_data_length(event);
S
Steven Rostedt 已提交
246 247 248 249 250 251 252
	default:
		BUG();
	}
	/* not hit */
	return 0;
}

253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269
/*
 * Return total length of time extend and data,
 *   or just the event length for all other events.
 */
static inline unsigned
rb_event_ts_length(struct ring_buffer_event *event)
{
	unsigned len = 0;

	if (event->type_len == RINGBUF_TYPE_TIME_EXTEND) {
		/* time extends include the data event after it */
		len = RB_LEN_TIME_EXTEND;
		event = skip_time_extend(event);
	}
	return len + rb_event_length(event);
}

S
Steven Rostedt 已提交
270 271 272
/**
 * ring_buffer_event_length - return the length of the event
 * @event: the event to get the length of
273 274 275 276 277 278
 *
 * Returns the size of the data load of a data event.
 * If the event is something other than a data event, it
 * returns the size of the event itself. With the exception
 * of a TIME EXTEND, where it still returns the size of the
 * data load of the data event after it.
S
Steven Rostedt 已提交
279 280 281
 */
unsigned ring_buffer_event_length(struct ring_buffer_event *event)
{
282 283 284 285 286 287
	unsigned length;

	if (event->type_len == RINGBUF_TYPE_TIME_EXTEND)
		event = skip_time_extend(event);

	length = rb_event_length(event);
288
	if (event->type_len > RINGBUF_TYPE_DATA_TYPE_LEN_MAX)
289 290 291 292 293
		return length;
	length -= RB_EVNT_HDR_SIZE;
	if (length > RB_MAX_SMALL_DATA + sizeof(event->array[0]))
                length -= sizeof(event->array[0]);
	return length;
S
Steven Rostedt 已提交
294
}
295
EXPORT_SYMBOL_GPL(ring_buffer_event_length);
S
Steven Rostedt 已提交
296 297

/* inline for ring buffer fast paths */
298
static void *
S
Steven Rostedt 已提交
299 300
rb_event_data(struct ring_buffer_event *event)
{
301 302
	if (event->type_len == RINGBUF_TYPE_TIME_EXTEND)
		event = skip_time_extend(event);
303
	BUG_ON(event->type_len > RINGBUF_TYPE_DATA_TYPE_LEN_MAX);
S
Steven Rostedt 已提交
304
	/* If length is in len field, then array[0] has the data */
305
	if (event->type_len)
S
Steven Rostedt 已提交
306 307 308 309 310 311 312 313 314 315 316 317 318
		return (void *)&event->array[0];
	/* Otherwise length is in array[0] and array[1] has the data */
	return (void *)&event->array[1];
}

/**
 * ring_buffer_event_data - return the data of the event
 * @event: the event to get the data from
 */
void *ring_buffer_event_data(struct ring_buffer_event *event)
{
	return rb_event_data(event);
}
319
EXPORT_SYMBOL_GPL(ring_buffer_event_data);
S
Steven Rostedt 已提交
320 321

#define for_each_buffer_cpu(buffer, cpu)		\
322
	for_each_cpu(cpu, buffer->cpumask)
S
Steven Rostedt 已提交
323 324 325 326 327

#define TS_SHIFT	27
#define TS_MASK		((1ULL << TS_SHIFT) - 1)
#define TS_DELTA_TEST	(~TS_MASK)

328 329
/* Flag when events were overwritten */
#define RB_MISSED_EVENTS	(1 << 31)
330 331
/* Missed count stored at end */
#define RB_MISSED_STORED	(1 << 30)
332

333
struct buffer_data_page {
334
	u64		 time_stamp;	/* page time stamp */
W
Wenji Huang 已提交
335
	local_t		 commit;	/* write committed index */
336 337 338
	unsigned char	 data[];	/* data of buffer page */
};

S
Steven Rostedt 已提交
339 340 341 342 343 344 345 346
/*
 * Note, the buffer_page list must be first. The buffer pages
 * are allocated in cache lines, which means that each buffer
 * page will be at the beginning of a cache line, and thus
 * the least significant bits will be zero. We use this to
 * add flags in the list struct pointers, to make the ring buffer
 * lockless.
 */
347
struct buffer_page {
348
	struct list_head list;		/* list of buffer pages */
349
	local_t		 write;		/* index for next write */
350
	unsigned	 read;		/* index for next read */
351
	local_t		 entries;	/* entries on this page */
352
	unsigned long	 real_end;	/* real end of data */
353
	struct buffer_data_page *page;	/* Actual data page */
S
Steven Rostedt 已提交
354 355
};

S
Steven Rostedt 已提交
356 357 358 359 360 361 362 363 364 365 366 367 368 369 370
/*
 * The buffer page counters, write and entries, must be reset
 * atomically when crossing page boundaries. To synchronize this
 * update, two counters are inserted into the number. One is
 * the actual counter for the write position or count on the page.
 *
 * The other is a counter of updaters. Before an update happens
 * the update partition of the counter is incremented. This will
 * allow the updater to update the counter atomically.
 *
 * The counter is 20 bits, and the state data is 12.
 */
#define RB_WRITE_MASK		0xfffff
#define RB_WRITE_INTCNT		(1 << 20)

371
static void rb_init_page(struct buffer_data_page *bpage)
372
{
373
	local_set(&bpage->commit, 0);
374 375
}

376 377 378 379 380 381
/**
 * ring_buffer_page_len - the size of data on the page.
 * @page: The page to read
 *
 * Returns the amount of data on the page, including buffer page header.
 */
382 383
size_t ring_buffer_page_len(void *page)
{
384 385
	return local_read(&((struct buffer_data_page *)page)->commit)
		+ BUF_PAGE_HDR_SIZE;
386 387
}

388 389 390 391
/*
 * Also stolen from mm/slob.c. Thanks to Mathieu Desnoyers for pointing
 * this issue out.
 */
392
static void free_buffer_page(struct buffer_page *bpage)
393
{
394
	free_page((unsigned long)bpage->page);
395
	kfree(bpage);
396 397
}

S
Steven Rostedt 已提交
398 399 400 401 402 403 404 405 406 407
/*
 * We need to fit the time_stamp delta into 27 bits.
 */
static inline int test_time_stamp(u64 delta)
{
	if (delta & TS_DELTA_TEST)
		return 1;
	return 0;
}

408
#define BUF_PAGE_SIZE (PAGE_SIZE - BUF_PAGE_HDR_SIZE)
S
Steven Rostedt 已提交
409

410 411 412
/* Max payload is BUF_PAGE_SIZE - header (8bytes) */
#define BUF_MAX_DATA_SIZE (BUF_PAGE_SIZE - (sizeof(u32) * 2))

413 414 415 416 417 418
int ring_buffer_print_page_header(struct trace_seq *s)
{
	struct buffer_data_page field;
	int ret;

	ret = trace_seq_printf(s, "\tfield: u64 timestamp;\t"
419 420 421
			       "offset:0;\tsize:%u;\tsigned:%u;\n",
			       (unsigned int)sizeof(field.time_stamp),
			       (unsigned int)is_signed_type(u64));
422 423

	ret = trace_seq_printf(s, "\tfield: local_t commit;\t"
424
			       "offset:%u;\tsize:%u;\tsigned:%u;\n",
425
			       (unsigned int)offsetof(typeof(field), commit),
426 427
			       (unsigned int)sizeof(field.commit),
			       (unsigned int)is_signed_type(long));
428

429 430 431 432 433 434
	ret = trace_seq_printf(s, "\tfield: int overwrite;\t"
			       "offset:%u;\tsize:%u;\tsigned:%u;\n",
			       (unsigned int)offsetof(typeof(field), commit),
			       1,
			       (unsigned int)is_signed_type(long));

435
	ret = trace_seq_printf(s, "\tfield: char data;\t"
436
			       "offset:%u;\tsize:%u;\tsigned:%u;\n",
437
			       (unsigned int)offsetof(typeof(field), data),
438 439
			       (unsigned int)BUF_PAGE_SIZE,
			       (unsigned int)is_signed_type(char));
440 441 442 443

	return ret;
}

S
Steven Rostedt 已提交
444 445 446 447 448
/*
 * head_page == tail_page && head == tail then buffer is empty.
 */
struct ring_buffer_per_cpu {
	int				cpu;
449
	atomic_t			record_disabled;
S
Steven Rostedt 已提交
450
	struct ring_buffer		*buffer;
451
	raw_spinlock_t			reader_lock;	/* serialize readers */
452
	arch_spinlock_t			lock;
S
Steven Rostedt 已提交
453
	struct lock_class_key		lock_key;
454
	unsigned int			nr_pages;
455
	struct list_head		*pages;
456 457
	struct buffer_page		*head_page;	/* read from head */
	struct buffer_page		*tail_page;	/* write to tail */
W
Wenji Huang 已提交
458
	struct buffer_page		*commit_page;	/* committed pages */
459
	struct buffer_page		*reader_page;
460 461
	unsigned long			lost_events;
	unsigned long			last_overrun;
462
	local_t				entries_bytes;
S
Steven Rostedt 已提交
463 464
	local_t				commit_overrun;
	local_t				overrun;
465
	local_t				entries;
466 467
	local_t				committing;
	local_t				commits;
S
Steven Rostedt 已提交
468
	unsigned long			read;
469
	unsigned long			read_bytes;
S
Steven Rostedt 已提交
470 471
	u64				write_stamp;
	u64				read_stamp;
472 473 474
	/* ring buffer pages to update, > 0 to add, < 0 to remove */
	int				nr_pages_to_update;
	struct list_head		new_pages; /* new pages to add */
475
	struct work_struct		update_pages_work;
476
	struct completion		update_done;
S
Steven Rostedt 已提交
477 478 479 480 481 482
};

struct ring_buffer {
	unsigned			flags;
	int				cpus;
	atomic_t			record_disabled;
483
	atomic_t			resize_disabled;
484
	cpumask_var_t			cpumask;
S
Steven Rostedt 已提交
485

486 487
	struct lock_class_key		*reader_lock_key;

S
Steven Rostedt 已提交
488 489 490
	struct mutex			mutex;

	struct ring_buffer_per_cpu	**buffers;
491

492
#ifdef CONFIG_HOTPLUG_CPU
493 494
	struct notifier_block		cpu_notify;
#endif
495
	u64				(*clock)(void);
S
Steven Rostedt 已提交
496 497 498 499 500 501
};

struct ring_buffer_iter {
	struct ring_buffer_per_cpu	*cpu_buffer;
	unsigned long			head;
	struct buffer_page		*head_page;
502 503
	struct buffer_page		*cache_reader_page;
	unsigned long			cache_read;
S
Steven Rostedt 已提交
504 505 506
	u64				read_stamp;
};

507
/* buffer may be either ring_buffer or ring_buffer_per_cpu */
508 509 510 511 512 513 514 515 516 517 518 519 520
#define RB_WARN_ON(b, cond)						\
	({								\
		int _____ret = unlikely(cond);				\
		if (_____ret) {						\
			if (__same_type(*(b), struct ring_buffer_per_cpu)) { \
				struct ring_buffer_per_cpu *__b =	\
					(void *)b;			\
				atomic_inc(&__b->buffer->record_disabled); \
			} else						\
				atomic_inc(&b->record_disabled);	\
			WARN_ON(1);					\
		}							\
		_____ret;						\
S
Steven Rostedt 已提交
521
	})
522

523 524 525
/* Up this if you want to test the TIME_EXTENTS and normalization */
#define DEBUG_SHIFT 0

526
static inline u64 rb_time_stamp(struct ring_buffer *buffer)
527 528 529 530 531
{
	/* shift to debug/test normalization and TIME_EXTENTS */
	return buffer->clock() << DEBUG_SHIFT;
}

532 533 534 535 536
u64 ring_buffer_time_stamp(struct ring_buffer *buffer, int cpu)
{
	u64 time;

	preempt_disable_notrace();
537
	time = rb_time_stamp(buffer);
538 539 540 541 542 543 544 545 546 547 548 549 550 551
	preempt_enable_no_resched_notrace();

	return time;
}
EXPORT_SYMBOL_GPL(ring_buffer_time_stamp);

void ring_buffer_normalize_time_stamp(struct ring_buffer *buffer,
				      int cpu, u64 *ts)
{
	/* Just stupid testing the normalize function and deltas */
	*ts >>= DEBUG_SHIFT;
}
EXPORT_SYMBOL_GPL(ring_buffer_normalize_time_stamp);

S
Steven Rostedt 已提交
552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641
/*
 * Making the ring buffer lockless makes things tricky.
 * Although writes only happen on the CPU that they are on,
 * and they only need to worry about interrupts. Reads can
 * happen on any CPU.
 *
 * The reader page is always off the ring buffer, but when the
 * reader finishes with a page, it needs to swap its page with
 * a new one from the buffer. The reader needs to take from
 * the head (writes go to the tail). But if a writer is in overwrite
 * mode and wraps, it must push the head page forward.
 *
 * Here lies the problem.
 *
 * The reader must be careful to replace only the head page, and
 * not another one. As described at the top of the file in the
 * ASCII art, the reader sets its old page to point to the next
 * page after head. It then sets the page after head to point to
 * the old reader page. But if the writer moves the head page
 * during this operation, the reader could end up with the tail.
 *
 * We use cmpxchg to help prevent this race. We also do something
 * special with the page before head. We set the LSB to 1.
 *
 * When the writer must push the page forward, it will clear the
 * bit that points to the head page, move the head, and then set
 * the bit that points to the new head page.
 *
 * We also don't want an interrupt coming in and moving the head
 * page on another writer. Thus we use the second LSB to catch
 * that too. Thus:
 *
 * head->list->prev->next        bit 1          bit 0
 *                              -------        -------
 * Normal page                     0              0
 * Points to head page             0              1
 * New head page                   1              0
 *
 * Note we can not trust the prev pointer of the head page, because:
 *
 * +----+       +-----+        +-----+
 * |    |------>|  T  |---X--->|  N  |
 * |    |<------|     |        |     |
 * +----+       +-----+        +-----+
 *   ^                           ^ |
 *   |          +-----+          | |
 *   +----------|  R  |----------+ |
 *              |     |<-----------+
 *              +-----+
 *
 * Key:  ---X-->  HEAD flag set in pointer
 *         T      Tail page
 *         R      Reader page
 *         N      Next page
 *
 * (see __rb_reserve_next() to see where this happens)
 *
 *  What the above shows is that the reader just swapped out
 *  the reader page with a page in the buffer, but before it
 *  could make the new header point back to the new page added
 *  it was preempted by a writer. The writer moved forward onto
 *  the new page added by the reader and is about to move forward
 *  again.
 *
 *  You can see, it is legitimate for the previous pointer of
 *  the head (or any page) not to point back to itself. But only
 *  temporarially.
 */

#define RB_PAGE_NORMAL		0UL
#define RB_PAGE_HEAD		1UL
#define RB_PAGE_UPDATE		2UL


#define RB_FLAG_MASK		3UL

/* PAGE_MOVED is not part of the mask */
#define RB_PAGE_MOVED		4UL

/*
 * rb_list_head - remove any bit
 */
static struct list_head *rb_list_head(struct list_head *list)
{
	unsigned long val = (unsigned long)list;

	return (struct list_head *)(val & ~RB_FLAG_MASK);
}

/*
642
 * rb_is_head_page - test if the given page is the head page
S
Steven Rostedt 已提交
643 644 645 646 647 648
 *
 * Because the reader may move the head_page pointer, we can
 * not trust what the head page is (it may be pointing to
 * the reader page). But if the next page is a header page,
 * its flags will be non zero.
 */
649
static inline int
S
Steven Rostedt 已提交
650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741
rb_is_head_page(struct ring_buffer_per_cpu *cpu_buffer,
		struct buffer_page *page, struct list_head *list)
{
	unsigned long val;

	val = (unsigned long)list->next;

	if ((val & ~RB_FLAG_MASK) != (unsigned long)&page->list)
		return RB_PAGE_MOVED;

	return val & RB_FLAG_MASK;
}

/*
 * rb_is_reader_page
 *
 * The unique thing about the reader page, is that, if the
 * writer is ever on it, the previous pointer never points
 * back to the reader page.
 */
static int rb_is_reader_page(struct buffer_page *page)
{
	struct list_head *list = page->list.prev;

	return rb_list_head(list->next) != &page->list;
}

/*
 * rb_set_list_to_head - set a list_head to be pointing to head.
 */
static void rb_set_list_to_head(struct ring_buffer_per_cpu *cpu_buffer,
				struct list_head *list)
{
	unsigned long *ptr;

	ptr = (unsigned long *)&list->next;
	*ptr |= RB_PAGE_HEAD;
	*ptr &= ~RB_PAGE_UPDATE;
}

/*
 * rb_head_page_activate - sets up head page
 */
static void rb_head_page_activate(struct ring_buffer_per_cpu *cpu_buffer)
{
	struct buffer_page *head;

	head = cpu_buffer->head_page;
	if (!head)
		return;

	/*
	 * Set the previous list pointer to have the HEAD flag.
	 */
	rb_set_list_to_head(cpu_buffer, head->list.prev);
}

static void rb_list_head_clear(struct list_head *list)
{
	unsigned long *ptr = (unsigned long *)&list->next;

	*ptr &= ~RB_FLAG_MASK;
}

/*
 * rb_head_page_dactivate - clears head page ptr (for free list)
 */
static void
rb_head_page_deactivate(struct ring_buffer_per_cpu *cpu_buffer)
{
	struct list_head *hd;

	/* Go through the whole list and clear any pointers found. */
	rb_list_head_clear(cpu_buffer->pages);

	list_for_each(hd, cpu_buffer->pages)
		rb_list_head_clear(hd);
}

static int rb_head_page_set(struct ring_buffer_per_cpu *cpu_buffer,
			    struct buffer_page *head,
			    struct buffer_page *prev,
			    int old_flag, int new_flag)
{
	struct list_head *list;
	unsigned long val = (unsigned long)&head->list;
	unsigned long ret;

	list = &prev->list;

	val &= ~RB_FLAG_MASK;

742 743
	ret = cmpxchg((unsigned long *)&list->next,
		      val | old_flag, val | new_flag);
S
Steven Rostedt 已提交
744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834

	/* check if the reader took the page */
	if ((ret & ~RB_FLAG_MASK) != val)
		return RB_PAGE_MOVED;

	return ret & RB_FLAG_MASK;
}

static int rb_head_page_set_update(struct ring_buffer_per_cpu *cpu_buffer,
				   struct buffer_page *head,
				   struct buffer_page *prev,
				   int old_flag)
{
	return rb_head_page_set(cpu_buffer, head, prev,
				old_flag, RB_PAGE_UPDATE);
}

static int rb_head_page_set_head(struct ring_buffer_per_cpu *cpu_buffer,
				 struct buffer_page *head,
				 struct buffer_page *prev,
				 int old_flag)
{
	return rb_head_page_set(cpu_buffer, head, prev,
				old_flag, RB_PAGE_HEAD);
}

static int rb_head_page_set_normal(struct ring_buffer_per_cpu *cpu_buffer,
				   struct buffer_page *head,
				   struct buffer_page *prev,
				   int old_flag)
{
	return rb_head_page_set(cpu_buffer, head, prev,
				old_flag, RB_PAGE_NORMAL);
}

static inline void rb_inc_page(struct ring_buffer_per_cpu *cpu_buffer,
			       struct buffer_page **bpage)
{
	struct list_head *p = rb_list_head((*bpage)->list.next);

	*bpage = list_entry(p, struct buffer_page, list);
}

static struct buffer_page *
rb_set_head_page(struct ring_buffer_per_cpu *cpu_buffer)
{
	struct buffer_page *head;
	struct buffer_page *page;
	struct list_head *list;
	int i;

	if (RB_WARN_ON(cpu_buffer, !cpu_buffer->head_page))
		return NULL;

	/* sanity check */
	list = cpu_buffer->pages;
	if (RB_WARN_ON(cpu_buffer, rb_list_head(list->prev->next) != list))
		return NULL;

	page = head = cpu_buffer->head_page;
	/*
	 * It is possible that the writer moves the header behind
	 * where we started, and we miss in one loop.
	 * A second loop should grab the header, but we'll do
	 * three loops just because I'm paranoid.
	 */
	for (i = 0; i < 3; i++) {
		do {
			if (rb_is_head_page(cpu_buffer, page, page->list.prev)) {
				cpu_buffer->head_page = page;
				return page;
			}
			rb_inc_page(cpu_buffer, &page);
		} while (page != head);
	}

	RB_WARN_ON(cpu_buffer, 1);

	return NULL;
}

static int rb_head_page_replace(struct buffer_page *old,
				struct buffer_page *new)
{
	unsigned long *ptr = (unsigned long *)&old->list.prev->next;
	unsigned long val;
	unsigned long ret;

	val = *ptr & ~RB_FLAG_MASK;
	val |= RB_PAGE_HEAD;

835
	ret = cmpxchg(ptr, val, (unsigned long)&new->list);
S
Steven Rostedt 已提交
836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885

	return ret == val;
}

/*
 * rb_tail_page_update - move the tail page forward
 *
 * Returns 1 if moved tail page, 0 if someone else did.
 */
static int rb_tail_page_update(struct ring_buffer_per_cpu *cpu_buffer,
			       struct buffer_page *tail_page,
			       struct buffer_page *next_page)
{
	struct buffer_page *old_tail;
	unsigned long old_entries;
	unsigned long old_write;
	int ret = 0;

	/*
	 * The tail page now needs to be moved forward.
	 *
	 * We need to reset the tail page, but without messing
	 * with possible erasing of data brought in by interrupts
	 * that have moved the tail page and are currently on it.
	 *
	 * We add a counter to the write field to denote this.
	 */
	old_write = local_add_return(RB_WRITE_INTCNT, &next_page->write);
	old_entries = local_add_return(RB_WRITE_INTCNT, &next_page->entries);

	/*
	 * Just make sure we have seen our old_write and synchronize
	 * with any interrupts that come in.
	 */
	barrier();

	/*
	 * If the tail page is still the same as what we think
	 * it is, then it is up to us to update the tail
	 * pointer.
	 */
	if (tail_page == cpu_buffer->tail_page) {
		/* Zero the write counter */
		unsigned long val = old_write & ~RB_WRITE_MASK;
		unsigned long eval = old_entries & ~RB_WRITE_MASK;

		/*
		 * This will only succeed if an interrupt did
		 * not come in and change it. In which case, we
		 * do not want to modify it.
886 887 888 889 890
		 *
		 * We add (void) to let the compiler know that we do not care
		 * about the return value of these functions. We use the
		 * cmpxchg to only update if an interrupt did not already
		 * do it for us. If the cmpxchg fails, we don't care.
S
Steven Rostedt 已提交
891
		 */
892 893
		(void)local_cmpxchg(&next_page->write, old_write, val);
		(void)local_cmpxchg(&next_page->entries, old_entries, eval);
S
Steven Rostedt 已提交
894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935

		/*
		 * No need to worry about races with clearing out the commit.
		 * it only can increment when a commit takes place. But that
		 * only happens in the outer most nested commit.
		 */
		local_set(&next_page->page->commit, 0);

		old_tail = cmpxchg(&cpu_buffer->tail_page,
				   tail_page, next_page);

		if (old_tail == tail_page)
			ret = 1;
	}

	return ret;
}

static int rb_check_bpage(struct ring_buffer_per_cpu *cpu_buffer,
			  struct buffer_page *bpage)
{
	unsigned long val = (unsigned long)bpage;

	if (RB_WARN_ON(cpu_buffer, val & RB_FLAG_MASK))
		return 1;

	return 0;
}

/**
 * rb_check_list - make sure a pointer to a list has the last bits zero
 */
static int rb_check_list(struct ring_buffer_per_cpu *cpu_buffer,
			 struct list_head *list)
{
	if (RB_WARN_ON(cpu_buffer, rb_list_head(list->prev) != list->prev))
		return 1;
	if (RB_WARN_ON(cpu_buffer, rb_list_head(list->next) != list->next))
		return 1;
	return 0;
}

S
Steven Rostedt 已提交
936 937 938 939
/**
 * check_pages - integrity check of buffer pages
 * @cpu_buffer: CPU buffer with pages to test
 *
W
Wenji Huang 已提交
940
 * As a safety measure we check to make sure the data pages have not
S
Steven Rostedt 已提交
941 942 943 944
 * been corrupted.
 */
static int rb_check_pages(struct ring_buffer_per_cpu *cpu_buffer)
{
945
	struct list_head *head = cpu_buffer->pages;
946
	struct buffer_page *bpage, *tmp;
S
Steven Rostedt 已提交
947

948 949 950 951
	/* Reset the head page if it exists */
	if (cpu_buffer->head_page)
		rb_set_head_page(cpu_buffer);

S
Steven Rostedt 已提交
952 953
	rb_head_page_deactivate(cpu_buffer);

S
Steven Rostedt 已提交
954 955 956 957
	if (RB_WARN_ON(cpu_buffer, head->next->prev != head))
		return -1;
	if (RB_WARN_ON(cpu_buffer, head->prev->next != head))
		return -1;
S
Steven Rostedt 已提交
958

S
Steven Rostedt 已提交
959 960 961
	if (rb_check_list(cpu_buffer, head))
		return -1;

962
	list_for_each_entry_safe(bpage, tmp, head, list) {
S
Steven Rostedt 已提交
963
		if (RB_WARN_ON(cpu_buffer,
964
			       bpage->list.next->prev != &bpage->list))
S
Steven Rostedt 已提交
965 966
			return -1;
		if (RB_WARN_ON(cpu_buffer,
967
			       bpage->list.prev->next != &bpage->list))
S
Steven Rostedt 已提交
968
			return -1;
S
Steven Rostedt 已提交
969 970
		if (rb_check_list(cpu_buffer, &bpage->list))
			return -1;
S
Steven Rostedt 已提交
971 972
	}

S
Steven Rostedt 已提交
973 974
	rb_head_page_activate(cpu_buffer);

S
Steven Rostedt 已提交
975 976 977
	return 0;
}

978
static int __rb_allocate_pages(int nr_pages, struct list_head *pages, int cpu)
S
Steven Rostedt 已提交
979
{
980
	int i;
981
	struct buffer_page *bpage, *tmp;
982

S
Steven Rostedt 已提交
983
	for (i = 0; i < nr_pages; i++) {
984
		struct page *page;
985 986 987 988 989
		/*
		 * __GFP_NORETRY flag makes sure that the allocation fails
		 * gracefully without invoking oom-killer and the system is
		 * not destabilized.
		 */
990
		bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),
991
				    GFP_KERNEL | __GFP_NORETRY,
992
				    cpu_to_node(cpu));
993
		if (!bpage)
994
			goto free_pages;
S
Steven Rostedt 已提交
995

996
		list_add(&bpage->list, pages);
S
Steven Rostedt 已提交
997

998
		page = alloc_pages_node(cpu_to_node(cpu),
999
					GFP_KERNEL | __GFP_NORETRY, 0);
1000
		if (!page)
S
Steven Rostedt 已提交
1001
			goto free_pages;
1002
		bpage->page = page_address(page);
1003
		rb_init_page(bpage->page);
S
Steven Rostedt 已提交
1004 1005
	}

1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026
	return 0;

free_pages:
	list_for_each_entry_safe(bpage, tmp, pages, list) {
		list_del_init(&bpage->list);
		free_buffer_page(bpage);
	}

	return -ENOMEM;
}

static int rb_allocate_pages(struct ring_buffer_per_cpu *cpu_buffer,
			     unsigned nr_pages)
{
	LIST_HEAD(pages);

	WARN_ON(!nr_pages);

	if (__rb_allocate_pages(nr_pages, &pages, cpu_buffer->cpu))
		return -ENOMEM;

1027 1028 1029 1030 1031 1032 1033
	/*
	 * The ring buffer page list is a circular list that does not
	 * start and end with a list head. All page list items point to
	 * other pages.
	 */
	cpu_buffer->pages = pages.next;
	list_del(&pages);
S
Steven Rostedt 已提交
1034

1035 1036
	cpu_buffer->nr_pages = nr_pages;

S
Steven Rostedt 已提交
1037 1038 1039 1040 1041 1042
	rb_check_pages(cpu_buffer);

	return 0;
}

static struct ring_buffer_per_cpu *
1043
rb_allocate_cpu_buffer(struct ring_buffer *buffer, int nr_pages, int cpu)
S
Steven Rostedt 已提交
1044 1045
{
	struct ring_buffer_per_cpu *cpu_buffer;
1046
	struct buffer_page *bpage;
1047
	struct page *page;
S
Steven Rostedt 已提交
1048 1049 1050 1051 1052 1053 1054 1055 1056
	int ret;

	cpu_buffer = kzalloc_node(ALIGN(sizeof(*cpu_buffer), cache_line_size()),
				  GFP_KERNEL, cpu_to_node(cpu));
	if (!cpu_buffer)
		return NULL;

	cpu_buffer->cpu = cpu;
	cpu_buffer->buffer = buffer;
1057
	raw_spin_lock_init(&cpu_buffer->reader_lock);
1058
	lockdep_set_class(&cpu_buffer->reader_lock, buffer->reader_lock_key);
1059
	cpu_buffer->lock = (arch_spinlock_t)__ARCH_SPIN_LOCK_UNLOCKED;
1060
	INIT_WORK(&cpu_buffer->update_pages_work, update_pages_handler);
1061
	init_completion(&cpu_buffer->update_done);
S
Steven Rostedt 已提交
1062

1063
	bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),
1064
			    GFP_KERNEL, cpu_to_node(cpu));
1065
	if (!bpage)
1066 1067
		goto fail_free_buffer;

S
Steven Rostedt 已提交
1068 1069
	rb_check_bpage(cpu_buffer, bpage);

1070
	cpu_buffer->reader_page = bpage;
1071 1072
	page = alloc_pages_node(cpu_to_node(cpu), GFP_KERNEL, 0);
	if (!page)
1073
		goto fail_free_reader;
1074
	bpage->page = page_address(page);
1075
	rb_init_page(bpage->page);
1076

1077
	INIT_LIST_HEAD(&cpu_buffer->reader_page->list);
1078
	INIT_LIST_HEAD(&cpu_buffer->new_pages);
1079

1080
	ret = rb_allocate_pages(cpu_buffer, nr_pages);
S
Steven Rostedt 已提交
1081
	if (ret < 0)
1082
		goto fail_free_reader;
S
Steven Rostedt 已提交
1083 1084

	cpu_buffer->head_page
1085
		= list_entry(cpu_buffer->pages, struct buffer_page, list);
S
Steven Rostedt 已提交
1086
	cpu_buffer->tail_page = cpu_buffer->commit_page = cpu_buffer->head_page;
S
Steven Rostedt 已提交
1087

S
Steven Rostedt 已提交
1088 1089
	rb_head_page_activate(cpu_buffer);

S
Steven Rostedt 已提交
1090 1091
	return cpu_buffer;

1092 1093 1094
 fail_free_reader:
	free_buffer_page(cpu_buffer->reader_page);

S
Steven Rostedt 已提交
1095 1096 1097 1098 1099 1100 1101
 fail_free_buffer:
	kfree(cpu_buffer);
	return NULL;
}

static void rb_free_cpu_buffer(struct ring_buffer_per_cpu *cpu_buffer)
{
1102
	struct list_head *head = cpu_buffer->pages;
1103
	struct buffer_page *bpage, *tmp;
S
Steven Rostedt 已提交
1104

1105 1106
	free_buffer_page(cpu_buffer->reader_page);

S
Steven Rostedt 已提交
1107 1108
	rb_head_page_deactivate(cpu_buffer);

1109 1110 1111 1112 1113 1114
	if (head) {
		list_for_each_entry_safe(bpage, tmp, head, list) {
			list_del_init(&bpage->list);
			free_buffer_page(bpage);
		}
		bpage = list_entry(head, struct buffer_page, list);
1115
		free_buffer_page(bpage);
S
Steven Rostedt 已提交
1116
	}
1117

S
Steven Rostedt 已提交
1118 1119 1120
	kfree(cpu_buffer);
}

1121
#ifdef CONFIG_HOTPLUG_CPU
1122 1123
static int rb_cpu_notify(struct notifier_block *self,
			 unsigned long action, void *hcpu);
1124 1125
#endif

S
Steven Rostedt 已提交
1126 1127
/**
 * ring_buffer_alloc - allocate a new ring_buffer
1128
 * @size: the size in bytes per cpu that is needed.
S
Steven Rostedt 已提交
1129 1130 1131 1132 1133 1134 1135
 * @flags: attributes to set for the ring buffer.
 *
 * Currently the only flag that is available is the RB_FL_OVERWRITE
 * flag. This flag means that the buffer will overwrite old data
 * when the buffer wraps. If this flag is not set, the buffer will
 * drop data when the tail hits the head.
 */
1136 1137
struct ring_buffer *__ring_buffer_alloc(unsigned long size, unsigned flags,
					struct lock_class_key *key)
S
Steven Rostedt 已提交
1138 1139 1140
{
	struct ring_buffer *buffer;
	int bsize;
1141
	int cpu, nr_pages;
S
Steven Rostedt 已提交
1142 1143 1144 1145 1146 1147 1148

	/* keep it in its own cache line */
	buffer = kzalloc(ALIGN(sizeof(*buffer), cache_line_size()),
			 GFP_KERNEL);
	if (!buffer)
		return NULL;

1149 1150 1151
	if (!alloc_cpumask_var(&buffer->cpumask, GFP_KERNEL))
		goto fail_free_buffer;

1152
	nr_pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
S
Steven Rostedt 已提交
1153
	buffer->flags = flags;
1154
	buffer->clock = trace_clock_local;
1155
	buffer->reader_lock_key = key;
S
Steven Rostedt 已提交
1156 1157

	/* need at least two pages */
1158 1159
	if (nr_pages < 2)
		nr_pages = 2;
S
Steven Rostedt 已提交
1160

1161 1162 1163 1164 1165 1166
	/*
	 * In case of non-hotplug cpu, if the ring-buffer is allocated
	 * in early initcall, it will not be notified of secondary cpus.
	 * In that off case, we need to allocate for all possible cpus.
	 */
#ifdef CONFIG_HOTPLUG_CPU
1167 1168
	get_online_cpus();
	cpumask_copy(buffer->cpumask, cpu_online_mask);
1169 1170 1171
#else
	cpumask_copy(buffer->cpumask, cpu_possible_mask);
#endif
S
Steven Rostedt 已提交
1172 1173 1174 1175 1176 1177
	buffer->cpus = nr_cpu_ids;

	bsize = sizeof(void *) * nr_cpu_ids;
	buffer->buffers = kzalloc(ALIGN(bsize, cache_line_size()),
				  GFP_KERNEL);
	if (!buffer->buffers)
1178
		goto fail_free_cpumask;
S
Steven Rostedt 已提交
1179 1180 1181

	for_each_buffer_cpu(buffer, cpu) {
		buffer->buffers[cpu] =
1182
			rb_allocate_cpu_buffer(buffer, nr_pages, cpu);
S
Steven Rostedt 已提交
1183 1184 1185 1186
		if (!buffer->buffers[cpu])
			goto fail_free_buffers;
	}

1187
#ifdef CONFIG_HOTPLUG_CPU
1188 1189 1190 1191 1192 1193
	buffer->cpu_notify.notifier_call = rb_cpu_notify;
	buffer->cpu_notify.priority = 0;
	register_cpu_notifier(&buffer->cpu_notify);
#endif

	put_online_cpus();
S
Steven Rostedt 已提交
1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204
	mutex_init(&buffer->mutex);

	return buffer;

 fail_free_buffers:
	for_each_buffer_cpu(buffer, cpu) {
		if (buffer->buffers[cpu])
			rb_free_cpu_buffer(buffer->buffers[cpu]);
	}
	kfree(buffer->buffers);

1205 1206
 fail_free_cpumask:
	free_cpumask_var(buffer->cpumask);
1207
	put_online_cpus();
1208

S
Steven Rostedt 已提交
1209 1210 1211 1212
 fail_free_buffer:
	kfree(buffer);
	return NULL;
}
1213
EXPORT_SYMBOL_GPL(__ring_buffer_alloc);
S
Steven Rostedt 已提交
1214 1215 1216 1217 1218 1219 1220 1221 1222 1223

/**
 * ring_buffer_free - free a ring buffer.
 * @buffer: the buffer to free.
 */
void
ring_buffer_free(struct ring_buffer *buffer)
{
	int cpu;

1224 1225
	get_online_cpus();

1226
#ifdef CONFIG_HOTPLUG_CPU
1227 1228 1229
	unregister_cpu_notifier(&buffer->cpu_notify);
#endif

S
Steven Rostedt 已提交
1230 1231 1232
	for_each_buffer_cpu(buffer, cpu)
		rb_free_cpu_buffer(buffer->buffers[cpu]);

1233 1234
	put_online_cpus();

1235
	kfree(buffer->buffers);
1236 1237
	free_cpumask_var(buffer->cpumask);

S
Steven Rostedt 已提交
1238 1239
	kfree(buffer);
}
1240
EXPORT_SYMBOL_GPL(ring_buffer_free);
S
Steven Rostedt 已提交
1241

1242 1243 1244 1245 1246 1247
void ring_buffer_set_clock(struct ring_buffer *buffer,
			   u64 (*clock)(void))
{
	buffer->clock = clock;
}

S
Steven Rostedt 已提交
1248 1249
static void rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer);

1250 1251 1252 1253 1254 1255 1256 1257 1258 1259
static inline unsigned long rb_page_entries(struct buffer_page *bpage)
{
	return local_read(&bpage->entries) & RB_WRITE_MASK;
}

static inline unsigned long rb_page_write(struct buffer_page *bpage)
{
	return local_read(&bpage->write) & RB_WRITE_MASK;
}

1260
static int
1261
rb_remove_pages(struct ring_buffer_per_cpu *cpu_buffer, unsigned int nr_pages)
S
Steven Rostedt 已提交
1262
{
1263 1264 1265 1266 1267 1268 1269 1270
	struct list_head *tail_page, *to_remove, *next_page;
	struct buffer_page *to_remove_page, *tmp_iter_page;
	struct buffer_page *last_page, *first_page;
	unsigned int nr_removed;
	unsigned long head_bit;
	int page_entries;

	head_bit = 0;
S
Steven Rostedt 已提交
1271

1272
	raw_spin_lock_irq(&cpu_buffer->reader_lock);
1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283
	atomic_inc(&cpu_buffer->record_disabled);
	/*
	 * We don't race with the readers since we have acquired the reader
	 * lock. We also don't race with writers after disabling recording.
	 * This makes it easy to figure out the first and the last page to be
	 * removed from the list. We unlink all the pages in between including
	 * the first and last pages. This is done in a busy loop so that we
	 * lose the least number of traces.
	 * The pages are freed after we restart recording and unlock readers.
	 */
	tail_page = &cpu_buffer->tail_page->list;
S
Steven Rostedt 已提交
1284

1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299
	/*
	 * tail page might be on reader page, we remove the next page
	 * from the ring buffer
	 */
	if (cpu_buffer->tail_page == cpu_buffer->reader_page)
		tail_page = rb_list_head(tail_page->next);
	to_remove = tail_page;

	/* start of pages to remove */
	first_page = list_entry(rb_list_head(to_remove->next),
				struct buffer_page, list);

	for (nr_removed = 0; nr_removed < nr_pages; nr_removed++) {
		to_remove = rb_list_head(to_remove)->next;
		head_bit |= (unsigned long)to_remove & RB_PAGE_HEAD;
S
Steven Rostedt 已提交
1300 1301
	}

1302
	next_page = rb_list_head(to_remove)->next;
S
Steven Rostedt 已提交
1303

1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329
	/*
	 * Now we remove all pages between tail_page and next_page.
	 * Make sure that we have head_bit value preserved for the
	 * next page
	 */
	tail_page->next = (struct list_head *)((unsigned long)next_page |
						head_bit);
	next_page = rb_list_head(next_page);
	next_page->prev = tail_page;

	/* make sure pages points to a valid page in the ring buffer */
	cpu_buffer->pages = next_page;

	/* update head page */
	if (head_bit)
		cpu_buffer->head_page = list_entry(next_page,
						struct buffer_page, list);

	/*
	 * change read pointer to make sure any read iterators reset
	 * themselves
	 */
	cpu_buffer->read = 0;

	/* pages are removed, resume tracing and then free the pages */
	atomic_dec(&cpu_buffer->record_disabled);
1330
	raw_spin_unlock_irq(&cpu_buffer->reader_lock);
1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349

	RB_WARN_ON(cpu_buffer, list_empty(cpu_buffer->pages));

	/* last buffer page to remove */
	last_page = list_entry(rb_list_head(to_remove), struct buffer_page,
				list);
	tmp_iter_page = first_page;

	do {
		to_remove_page = tmp_iter_page;
		rb_inc_page(cpu_buffer, &tmp_iter_page);

		/* update the counters */
		page_entries = rb_page_entries(to_remove_page);
		if (page_entries) {
			/*
			 * If something was added to this page, it was full
			 * since it is not the tail page. So we deduct the
			 * bytes consumed in ring buffer from here.
1350
			 * Increment overrun to account for the lost events.
1351
			 */
1352
			local_add(page_entries, &cpu_buffer->overrun);
1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365
			local_sub(BUF_PAGE_SIZE, &cpu_buffer->entries_bytes);
		}

		/*
		 * We have already removed references to this list item, just
		 * free up the buffer_page and its page
		 */
		free_buffer_page(to_remove_page);
		nr_removed--;

	} while (to_remove_page != last_page);

	RB_WARN_ON(cpu_buffer, nr_removed);
1366 1367

	return nr_removed == 0;
S
Steven Rostedt 已提交
1368 1369
}

1370 1371
static int
rb_insert_pages(struct ring_buffer_per_cpu *cpu_buffer)
S
Steven Rostedt 已提交
1372
{
1373 1374
	struct list_head *pages = &cpu_buffer->new_pages;
	int retries, success;
S
Steven Rostedt 已提交
1375

1376
	raw_spin_lock_irq(&cpu_buffer->reader_lock);
1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396
	/*
	 * We are holding the reader lock, so the reader page won't be swapped
	 * in the ring buffer. Now we are racing with the writer trying to
	 * move head page and the tail page.
	 * We are going to adapt the reader page update process where:
	 * 1. We first splice the start and end of list of new pages between
	 *    the head page and its previous page.
	 * 2. We cmpxchg the prev_page->next to point from head page to the
	 *    start of new pages list.
	 * 3. Finally, we update the head->prev to the end of new list.
	 *
	 * We will try this process 10 times, to make sure that we don't keep
	 * spinning.
	 */
	retries = 10;
	success = 0;
	while (retries--) {
		struct list_head *head_page, *prev_page, *r;
		struct list_head *last_page, *first_page;
		struct list_head *head_page_with_bit;
S
Steven Rostedt 已提交
1397

1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421
		head_page = &rb_set_head_page(cpu_buffer)->list;
		prev_page = head_page->prev;

		first_page = pages->next;
		last_page  = pages->prev;

		head_page_with_bit = (struct list_head *)
				     ((unsigned long)head_page | RB_PAGE_HEAD);

		last_page->next = head_page_with_bit;
		first_page->prev = prev_page;

		r = cmpxchg(&prev_page->next, head_page_with_bit, first_page);

		if (r == head_page_with_bit) {
			/*
			 * yay, we replaced the page pointer to our new list,
			 * now, we just have to update to head page's prev
			 * pointer to point to end of list
			 */
			head_page->prev = last_page;
			success = 1;
			break;
		}
S
Steven Rostedt 已提交
1422 1423
	}

1424 1425 1426 1427 1428 1429 1430
	if (success)
		INIT_LIST_HEAD(pages);
	/*
	 * If we weren't successful in adding in new pages, warn and stop
	 * tracing
	 */
	RB_WARN_ON(cpu_buffer, !success);
1431
	raw_spin_unlock_irq(&cpu_buffer->reader_lock);
1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442

	/* free pages if they weren't inserted */
	if (!success) {
		struct buffer_page *bpage, *tmp;
		list_for_each_entry_safe(bpage, tmp, &cpu_buffer->new_pages,
					 list) {
			list_del_init(&bpage->list);
			free_buffer_page(bpage);
		}
	}
	return success;
S
Steven Rostedt 已提交
1443 1444
}

1445
static void rb_update_pages(struct ring_buffer_per_cpu *cpu_buffer)
1446
{
1447 1448
	int success;

1449
	if (cpu_buffer->nr_pages_to_update > 0)
1450
		success = rb_insert_pages(cpu_buffer);
1451
	else
1452 1453
		success = rb_remove_pages(cpu_buffer,
					-cpu_buffer->nr_pages_to_update);
1454

1455 1456
	if (success)
		cpu_buffer->nr_pages += cpu_buffer->nr_pages_to_update;
1457 1458 1459 1460 1461 1462 1463
}

static void update_pages_handler(struct work_struct *work)
{
	struct ring_buffer_per_cpu *cpu_buffer = container_of(work,
			struct ring_buffer_per_cpu, update_pages_work);
	rb_update_pages(cpu_buffer);
1464
	complete(&cpu_buffer->update_done);
1465 1466
}

S
Steven Rostedt 已提交
1467 1468 1469 1470 1471 1472 1473
/**
 * ring_buffer_resize - resize the ring buffer
 * @buffer: the buffer to resize.
 * @size: the new size.
 *
 * Minimum size is 2 * BUF_PAGE_SIZE.
 *
1474
 * Returns 0 on success and < 0 on failure.
S
Steven Rostedt 已提交
1475
 */
1476 1477
int ring_buffer_resize(struct ring_buffer *buffer, unsigned long size,
			int cpu_id)
S
Steven Rostedt 已提交
1478 1479
{
	struct ring_buffer_per_cpu *cpu_buffer;
1480
	unsigned nr_pages;
1481
	int cpu, err = 0;
S
Steven Rostedt 已提交
1482

1483 1484 1485 1486 1487 1488
	/*
	 * Always succeed at resizing a non-existent buffer:
	 */
	if (!buffer)
		return size;

1489 1490 1491 1492 1493
	/* Make sure the requested buffer exists */
	if (cpu_id != RING_BUFFER_ALL_CPUS &&
	    !cpumask_test_cpu(cpu_id, buffer->cpumask))
		return size;

S
Steven Rostedt 已提交
1494 1495 1496 1497 1498 1499 1500
	size = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
	size *= BUF_PAGE_SIZE;

	/* we need a minimum of two pages */
	if (size < BUF_PAGE_SIZE * 2)
		size = BUF_PAGE_SIZE * 2;

1501
	nr_pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
1502

1503 1504 1505 1506 1507 1508 1509
	/*
	 * Don't succeed if resizing is disabled, as a reader might be
	 * manipulating the ring buffer and is expecting a sane state while
	 * this is true.
	 */
	if (atomic_read(&buffer->resize_disabled))
		return -EBUSY;
1510

1511
	/* prevent another thread from changing buffer sizes */
S
Steven Rostedt 已提交
1512 1513
	mutex_lock(&buffer->mutex);

1514 1515
	if (cpu_id == RING_BUFFER_ALL_CPUS) {
		/* calculate the pages to update */
S
Steven Rostedt 已提交
1516 1517 1518
		for_each_buffer_cpu(buffer, cpu) {
			cpu_buffer = buffer->buffers[cpu];

1519 1520 1521 1522 1523 1524 1525
			cpu_buffer->nr_pages_to_update = nr_pages -
							cpu_buffer->nr_pages;
			/*
			 * nothing more to do for removing pages or no update
			 */
			if (cpu_buffer->nr_pages_to_update <= 0)
				continue;
1526
			/*
1527 1528
			 * to add pages, make sure all new pages can be
			 * allocated without receiving ENOMEM
1529
			 */
1530 1531
			INIT_LIST_HEAD(&cpu_buffer->new_pages);
			if (__rb_allocate_pages(cpu_buffer->nr_pages_to_update,
1532
						&cpu_buffer->new_pages, cpu)) {
1533
				/* not enough memory for new pages */
1534 1535 1536 1537 1538 1539 1540 1541
				err = -ENOMEM;
				goto out_err;
			}
		}

		get_online_cpus();
		/*
		 * Fire off all the required work handlers
1542
		 * We can't schedule on offline CPUs, but it's not necessary
1543 1544 1545 1546
		 * since we can change their buffer sizes without any race.
		 */
		for_each_buffer_cpu(buffer, cpu) {
			cpu_buffer = buffer->buffers[cpu];
1547
			if (!cpu_buffer->nr_pages_to_update)
1548 1549
				continue;

1550 1551 1552 1553 1554
			if (cpu_online(cpu))
				schedule_work_on(cpu,
						&cpu_buffer->update_pages_work);
			else
				rb_update_pages(cpu_buffer);
S
Steven Rostedt 已提交
1555 1556
		}

1557 1558 1559
		/* wait for all the updates to complete */
		for_each_buffer_cpu(buffer, cpu) {
			cpu_buffer = buffer->buffers[cpu];
1560
			if (!cpu_buffer->nr_pages_to_update)
1561 1562
				continue;

1563 1564
			if (cpu_online(cpu))
				wait_for_completion(&cpu_buffer->update_done);
1565
			cpu_buffer->nr_pages_to_update = 0;
1566
		}
1567 1568

		put_online_cpus();
1569
	} else {
1570 1571 1572 1573
		/* Make sure this CPU has been intitialized */
		if (!cpumask_test_cpu(cpu_id, buffer->cpumask))
			goto out;

1574
		cpu_buffer = buffer->buffers[cpu_id];
1575

1576 1577
		if (nr_pages == cpu_buffer->nr_pages)
			goto out;
S
Steven Rostedt 已提交
1578

1579 1580 1581 1582 1583 1584
		cpu_buffer->nr_pages_to_update = nr_pages -
						cpu_buffer->nr_pages;

		INIT_LIST_HEAD(&cpu_buffer->new_pages);
		if (cpu_buffer->nr_pages_to_update > 0 &&
			__rb_allocate_pages(cpu_buffer->nr_pages_to_update,
1585 1586 1587 1588
					    &cpu_buffer->new_pages, cpu_id)) {
			err = -ENOMEM;
			goto out_err;
		}
1589

1590 1591 1592 1593 1594
		get_online_cpus();

		if (cpu_online(cpu_id)) {
			schedule_work_on(cpu_id,
					 &cpu_buffer->update_pages_work);
1595
			wait_for_completion(&cpu_buffer->update_done);
1596 1597 1598 1599
		} else
			rb_update_pages(cpu_buffer);

		cpu_buffer->nr_pages_to_update = 0;
1600
		put_online_cpus();
1601
	}
S
Steven Rostedt 已提交
1602 1603

 out:
1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626
	/*
	 * The ring buffer resize can happen with the ring buffer
	 * enabled, so that the update disturbs the tracing as little
	 * as possible. But if the buffer is disabled, we do not need
	 * to worry about that, and we can take the time to verify
	 * that the buffer is not corrupt.
	 */
	if (atomic_read(&buffer->record_disabled)) {
		atomic_inc(&buffer->record_disabled);
		/*
		 * Even though the buffer was disabled, we must make sure
		 * that it is truly disabled before calling rb_check_pages.
		 * There could have been a race between checking
		 * record_disable and incrementing it.
		 */
		synchronize_sched();
		for_each_buffer_cpu(buffer, cpu) {
			cpu_buffer = buffer->buffers[cpu];
			rb_check_pages(cpu_buffer);
		}
		atomic_dec(&buffer->record_disabled);
	}

S
Steven Rostedt 已提交
1627 1628 1629
	mutex_unlock(&buffer->mutex);
	return size;

1630
 out_err:
1631 1632
	for_each_buffer_cpu(buffer, cpu) {
		struct buffer_page *bpage, *tmp;
1633

1634 1635
		cpu_buffer = buffer->buffers[cpu];
		cpu_buffer->nr_pages_to_update = 0;
1636

1637 1638
		if (list_empty(&cpu_buffer->new_pages))
			continue;
1639

1640 1641 1642 1643 1644
		list_for_each_entry_safe(bpage, tmp, &cpu_buffer->new_pages,
					list) {
			list_del_init(&bpage->list);
			free_buffer_page(bpage);
		}
S
Steven Rostedt 已提交
1645
	}
1646
	mutex_unlock(&buffer->mutex);
1647
	return err;
S
Steven Rostedt 已提交
1648
}
1649
EXPORT_SYMBOL_GPL(ring_buffer_resize);
S
Steven Rostedt 已提交
1650

1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661
void ring_buffer_change_overwrite(struct ring_buffer *buffer, int val)
{
	mutex_lock(&buffer->mutex);
	if (val)
		buffer->flags |= RB_FL_OVERWRITE;
	else
		buffer->flags &= ~RB_FL_OVERWRITE;
	mutex_unlock(&buffer->mutex);
}
EXPORT_SYMBOL_GPL(ring_buffer_change_overwrite);

S
Steven Rostedt 已提交
1662
static inline void *
1663
__rb_data_page_index(struct buffer_data_page *bpage, unsigned index)
S
Steven Rostedt 已提交
1664
{
1665
	return bpage->data + index;
S
Steven Rostedt 已提交
1666 1667
}

1668
static inline void *__rb_page_index(struct buffer_page *bpage, unsigned index)
S
Steven Rostedt 已提交
1669
{
1670
	return bpage->page->data + index;
S
Steven Rostedt 已提交
1671 1672 1673
}

static inline struct ring_buffer_event *
1674
rb_reader_event(struct ring_buffer_per_cpu *cpu_buffer)
S
Steven Rostedt 已提交
1675
{
1676 1677 1678 1679
	return __rb_page_index(cpu_buffer->reader_page,
			       cpu_buffer->reader_page->read);
}

S
Steven Rostedt 已提交
1680 1681 1682
static inline struct ring_buffer_event *
rb_iter_head_event(struct ring_buffer_iter *iter)
{
1683
	return __rb_page_index(iter->head_page, iter->head);
S
Steven Rostedt 已提交
1684 1685
}

S
Steven Rostedt 已提交
1686 1687
static inline unsigned rb_page_commit(struct buffer_page *bpage)
{
1688
	return local_read(&bpage->page->commit);
S
Steven Rostedt 已提交
1689 1690
}

L
Lucas De Marchi 已提交
1691
/* Size is determined by what has been committed */
S
Steven Rostedt 已提交
1692 1693 1694 1695 1696 1697 1698 1699 1700 1701 1702 1703 1704 1705 1706 1707
static inline unsigned rb_page_size(struct buffer_page *bpage)
{
	return rb_page_commit(bpage);
}

static inline unsigned
rb_commit_index(struct ring_buffer_per_cpu *cpu_buffer)
{
	return rb_page_commit(cpu_buffer->commit_page);
}

static inline unsigned
rb_event_index(struct ring_buffer_event *event)
{
	unsigned long addr = (unsigned long)event;

1708
	return (addr & ~PAGE_MASK) - BUF_PAGE_HDR_SIZE;
S
Steven Rostedt 已提交
1709 1710
}

1711
static inline int
1712 1713
rb_event_is_commit(struct ring_buffer_per_cpu *cpu_buffer,
		   struct ring_buffer_event *event)
S
Steven Rostedt 已提交
1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 1724
{
	unsigned long addr = (unsigned long)event;
	unsigned long index;

	index = rb_event_index(event);
	addr &= PAGE_MASK;

	return cpu_buffer->commit_page->page == (void *)addr &&
		rb_commit_index(cpu_buffer) == index;
}

1725
static void
S
Steven Rostedt 已提交
1726
rb_set_commit_to_write(struct ring_buffer_per_cpu *cpu_buffer)
S
Steven Rostedt 已提交
1727
{
S
Steven Rostedt 已提交
1728 1729
	unsigned long max_count;

S
Steven Rostedt 已提交
1730 1731 1732 1733 1734 1735 1736 1737
	/*
	 * We only race with interrupts and NMIs on this CPU.
	 * If we own the commit event, then we can commit
	 * all others that interrupted us, since the interruptions
	 * are in stack format (they finish before they come
	 * back to us). This allows us to do a simple loop to
	 * assign the commit to the tail.
	 */
1738
 again:
1739
	max_count = cpu_buffer->nr_pages * 100;
S
Steven Rostedt 已提交
1740

S
Steven Rostedt 已提交
1741
	while (cpu_buffer->commit_page != cpu_buffer->tail_page) {
S
Steven Rostedt 已提交
1742 1743 1744 1745 1746 1747 1748
		if (RB_WARN_ON(cpu_buffer, !(--max_count)))
			return;
		if (RB_WARN_ON(cpu_buffer,
			       rb_is_reader_page(cpu_buffer->tail_page)))
			return;
		local_set(&cpu_buffer->commit_page->page->commit,
			  rb_page_write(cpu_buffer->commit_page));
S
Steven Rostedt 已提交
1749
		rb_inc_page(cpu_buffer, &cpu_buffer->commit_page);
1750 1751
		cpu_buffer->write_stamp =
			cpu_buffer->commit_page->page->time_stamp;
S
Steven Rostedt 已提交
1752 1753 1754 1755 1756
		/* add barrier to keep gcc from optimizing too much */
		barrier();
	}
	while (rb_commit_index(cpu_buffer) !=
	       rb_page_write(cpu_buffer->commit_page)) {
S
Steven Rostedt 已提交
1757 1758 1759 1760 1761 1762

		local_set(&cpu_buffer->commit_page->page->commit,
			  rb_page_write(cpu_buffer->commit_page));
		RB_WARN_ON(cpu_buffer,
			   local_read(&cpu_buffer->commit_page->page->commit) &
			   ~RB_WRITE_MASK);
S
Steven Rostedt 已提交
1763 1764
		barrier();
	}
1765 1766 1767 1768 1769 1770 1771 1772 1773 1774 1775

	/* again, keep gcc from optimizing */
	barrier();

	/*
	 * If an interrupt came in just after the first while loop
	 * and pushed the tail page forward, we will be left with
	 * a dangling commit that will never go forward.
	 */
	if (unlikely(cpu_buffer->commit_page != cpu_buffer->tail_page))
		goto again;
S
Steven Rostedt 已提交
1776 1777
}

1778
static void rb_reset_reader_page(struct ring_buffer_per_cpu *cpu_buffer)
S
Steven Rostedt 已提交
1779
{
1780
	cpu_buffer->read_stamp = cpu_buffer->reader_page->page->time_stamp;
1781
	cpu_buffer->reader_page->read = 0;
1782 1783
}

1784
static void rb_inc_iter(struct ring_buffer_iter *iter)
1785 1786 1787 1788 1789 1790 1791 1792 1793 1794
{
	struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;

	/*
	 * The iterator could be on the reader page (it starts there).
	 * But the head could have moved, since the reader was
	 * found. Check for this case and assign the iterator
	 * to the head page instead of next.
	 */
	if (iter->head_page == cpu_buffer->reader_page)
S
Steven Rostedt 已提交
1795
		iter->head_page = rb_set_head_page(cpu_buffer);
1796 1797 1798
	else
		rb_inc_page(cpu_buffer, &iter->head_page);

1799
	iter->read_stamp = iter->head_page->page->time_stamp;
S
Steven Rostedt 已提交
1800 1801 1802
	iter->head = 0;
}

1803 1804 1805 1806 1807 1808 1809 1810 1811 1812 1813 1814 1815 1816 1817 1818 1819 1820 1821
/* Slow path, do not inline */
static noinline struct ring_buffer_event *
rb_add_time_stamp(struct ring_buffer_event *event, u64 delta)
{
	event->type_len = RINGBUF_TYPE_TIME_EXTEND;

	/* Not the first event on the page? */
	if (rb_event_index(event)) {
		event->time_delta = delta & TS_MASK;
		event->array[0] = delta >> TS_SHIFT;
	} else {
		/* nope, just zero it */
		event->time_delta = 0;
		event->array[0] = 0;
	}

	return skip_time_extend(event);
}

S
Steven Rostedt 已提交
1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832
/**
 * ring_buffer_update_event - update event type and data
 * @event: the even to update
 * @type: the type of event
 * @length: the size of the event field in the ring buffer
 *
 * Update the type and data fields of the event. The length
 * is the actual size that is written to the ring buffer,
 * and with this, we can determine what to place into the
 * data field.
 */
1833
static void
1834 1835 1836
rb_update_event(struct ring_buffer_per_cpu *cpu_buffer,
		struct ring_buffer_event *event, unsigned length,
		int add_timestamp, u64 delta)
S
Steven Rostedt 已提交
1837
{
1838 1839 1840
	/* Only a commit updates the timestamp */
	if (unlikely(!rb_event_is_commit(cpu_buffer, event)))
		delta = 0;
S
Steven Rostedt 已提交
1841

1842 1843 1844 1845 1846 1847 1848 1849
	/*
	 * If we need to add a timestamp, then we
	 * add it to the start of the resevered space.
	 */
	if (unlikely(add_timestamp)) {
		event = rb_add_time_stamp(event, delta);
		length -= RB_LEN_TIME_EXTEND;
		delta = 0;
S
Steven Rostedt 已提交
1850
	}
1851 1852 1853 1854 1855 1856 1857 1858

	event->time_delta = delta;
	length -= RB_EVNT_HDR_SIZE;
	if (length > RB_MAX_SMALL_DATA || RB_FORCE_8BYTE_ALIGNMENT) {
		event->type_len = 0;
		event->array[0] = length;
	} else
		event->type_len = DIV_ROUND_UP(length, RB_ALIGNMENT);
S
Steven Rostedt 已提交
1859 1860
}

S
Steven Rostedt 已提交
1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877 1878 1879 1880 1881 1882 1883 1884 1885 1886 1887 1888 1889 1890 1891 1892 1893 1894 1895 1896 1897 1898 1899 1900 1901 1902 1903 1904 1905 1906
/*
 * rb_handle_head_page - writer hit the head page
 *
 * Returns: +1 to retry page
 *           0 to continue
 *          -1 on error
 */
static int
rb_handle_head_page(struct ring_buffer_per_cpu *cpu_buffer,
		    struct buffer_page *tail_page,
		    struct buffer_page *next_page)
{
	struct buffer_page *new_head;
	int entries;
	int type;
	int ret;

	entries = rb_page_entries(next_page);

	/*
	 * The hard part is here. We need to move the head
	 * forward, and protect against both readers on
	 * other CPUs and writers coming in via interrupts.
	 */
	type = rb_head_page_set_update(cpu_buffer, next_page, tail_page,
				       RB_PAGE_HEAD);

	/*
	 * type can be one of four:
	 *  NORMAL - an interrupt already moved it for us
	 *  HEAD   - we are the first to get here.
	 *  UPDATE - we are the interrupt interrupting
	 *           a current move.
	 *  MOVED  - a reader on another CPU moved the next
	 *           pointer to its reader page. Give up
	 *           and try again.
	 */

	switch (type) {
	case RB_PAGE_HEAD:
		/*
		 * We changed the head to UPDATE, thus
		 * it is our responsibility to update
		 * the counters.
		 */
		local_add(entries, &cpu_buffer->overrun);
1907
		local_sub(BUF_PAGE_SIZE, &cpu_buffer->entries_bytes);
S
Steven Rostedt 已提交
1908 1909 1910 1911 1912 1913 1914 1915 1916 1917 1918 1919 1920 1921 1922 1923 1924 1925 1926 1927 1928 1929 1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957 1958 1959 1960 1961 1962 1963 1964 1965 1966 1967 1968 1969 1970 1971 1972 1973 1974 1975 1976 1977 1978 1979 1980 1981 1982 1983 1984 1985 1986 1987 1988 1989 1990 1991 1992 1993 1994 1995 1996 1997 1998 1999 2000 2001 2002 2003 2004 2005 2006 2007 2008 2009 2010 2011 2012 2013 2014 2015 2016 2017 2018

		/*
		 * The entries will be zeroed out when we move the
		 * tail page.
		 */

		/* still more to do */
		break;

	case RB_PAGE_UPDATE:
		/*
		 * This is an interrupt that interrupt the
		 * previous update. Still more to do.
		 */
		break;
	case RB_PAGE_NORMAL:
		/*
		 * An interrupt came in before the update
		 * and processed this for us.
		 * Nothing left to do.
		 */
		return 1;
	case RB_PAGE_MOVED:
		/*
		 * The reader is on another CPU and just did
		 * a swap with our next_page.
		 * Try again.
		 */
		return 1;
	default:
		RB_WARN_ON(cpu_buffer, 1); /* WTF??? */
		return -1;
	}

	/*
	 * Now that we are here, the old head pointer is
	 * set to UPDATE. This will keep the reader from
	 * swapping the head page with the reader page.
	 * The reader (on another CPU) will spin till
	 * we are finished.
	 *
	 * We just need to protect against interrupts
	 * doing the job. We will set the next pointer
	 * to HEAD. After that, we set the old pointer
	 * to NORMAL, but only if it was HEAD before.
	 * otherwise we are an interrupt, and only
	 * want the outer most commit to reset it.
	 */
	new_head = next_page;
	rb_inc_page(cpu_buffer, &new_head);

	ret = rb_head_page_set_head(cpu_buffer, new_head, next_page,
				    RB_PAGE_NORMAL);

	/*
	 * Valid returns are:
	 *  HEAD   - an interrupt came in and already set it.
	 *  NORMAL - One of two things:
	 *            1) We really set it.
	 *            2) A bunch of interrupts came in and moved
	 *               the page forward again.
	 */
	switch (ret) {
	case RB_PAGE_HEAD:
	case RB_PAGE_NORMAL:
		/* OK */
		break;
	default:
		RB_WARN_ON(cpu_buffer, 1);
		return -1;
	}

	/*
	 * It is possible that an interrupt came in,
	 * set the head up, then more interrupts came in
	 * and moved it again. When we get back here,
	 * the page would have been set to NORMAL but we
	 * just set it back to HEAD.
	 *
	 * How do you detect this? Well, if that happened
	 * the tail page would have moved.
	 */
	if (ret == RB_PAGE_NORMAL) {
		/*
		 * If the tail had moved passed next, then we need
		 * to reset the pointer.
		 */
		if (cpu_buffer->tail_page != tail_page &&
		    cpu_buffer->tail_page != next_page)
			rb_head_page_set_normal(cpu_buffer, new_head,
						next_page,
						RB_PAGE_HEAD);
	}

	/*
	 * If this was the outer most commit (the one that
	 * changed the original pointer from HEAD to UPDATE),
	 * then it is up to us to reset it to NORMAL.
	 */
	if (type == RB_PAGE_HEAD) {
		ret = rb_head_page_set_normal(cpu_buffer, next_page,
					      tail_page,
					      RB_PAGE_UPDATE);
		if (RB_WARN_ON(cpu_buffer,
			       ret != RB_PAGE_UPDATE))
			return -1;
	}

	return 0;
}

2019
static unsigned rb_calculate_event_length(unsigned length)
S
Steven Rostedt 已提交
2020 2021 2022 2023 2024 2025 2026
{
	struct ring_buffer_event event; /* Used only for sizeof array */

	/* zero length can cause confusions */
	if (!length)
		length = 1;

2027
	if (length > RB_MAX_SMALL_DATA || RB_FORCE_8BYTE_ALIGNMENT)
S
Steven Rostedt 已提交
2028 2029 2030
		length += sizeof(event.array[0]);

	length += RB_EVNT_HDR_SIZE;
2031
	length = ALIGN(length, RB_ARCH_ALIGNMENT);
S
Steven Rostedt 已提交
2032 2033 2034 2035

	return length;
}

2036 2037 2038 2039 2040 2041 2042 2043 2044 2045 2046 2047
static inline void
rb_reset_tail(struct ring_buffer_per_cpu *cpu_buffer,
	      struct buffer_page *tail_page,
	      unsigned long tail, unsigned long length)
{
	struct ring_buffer_event *event;

	/*
	 * Only the event that crossed the page boundary
	 * must fill the old tail_page with padding.
	 */
	if (tail >= BUF_PAGE_SIZE) {
2048 2049 2050 2051 2052 2053 2054 2055
		/*
		 * If the page was filled, then we still need
		 * to update the real_end. Reset it to zero
		 * and the reader will ignore it.
		 */
		if (tail == BUF_PAGE_SIZE)
			tail_page->real_end = 0;

2056 2057 2058 2059 2060
		local_sub(length, &tail_page->write);
		return;
	}

	event = __rb_page_index(tail_page, tail);
2061
	kmemcheck_annotate_bitfield(event, bitfield);
2062

2063 2064 2065
	/* account for padding bytes */
	local_add(BUF_PAGE_SIZE - tail, &cpu_buffer->entries_bytes);

2066 2067 2068 2069 2070 2071 2072
	/*
	 * Save the original length to the meta data.
	 * This will be used by the reader to add lost event
	 * counter.
	 */
	tail_page->real_end = tail;

2073 2074 2075 2076 2077 2078 2079 2080 2081 2082 2083 2084 2085 2086 2087 2088 2089 2090 2091 2092 2093 2094 2095 2096 2097 2098 2099 2100 2101 2102 2103 2104
	/*
	 * If this event is bigger than the minimum size, then
	 * we need to be careful that we don't subtract the
	 * write counter enough to allow another writer to slip
	 * in on this page.
	 * We put in a discarded commit instead, to make sure
	 * that this space is not used again.
	 *
	 * If we are less than the minimum size, we don't need to
	 * worry about it.
	 */
	if (tail > (BUF_PAGE_SIZE - RB_EVNT_MIN_SIZE)) {
		/* No room for any events */

		/* Mark the rest of the page with padding */
		rb_event_set_padding(event);

		/* Set the write back to the previous setting */
		local_sub(length, &tail_page->write);
		return;
	}

	/* Put in a discarded event */
	event->array[0] = (BUF_PAGE_SIZE - tail) - RB_EVNT_HDR_SIZE;
	event->type_len = RINGBUF_TYPE_PADDING;
	/* time delta must be non zero */
	event->time_delta = 1;

	/* Set write to end of buffer */
	length = (tail + length) - BUF_PAGE_SIZE;
	local_sub(length, &tail_page->write);
}
2105

2106 2107 2108 2109
/*
 * This is the slow path, force gcc not to inline it.
 */
static noinline struct ring_buffer_event *
2110 2111
rb_move_tail(struct ring_buffer_per_cpu *cpu_buffer,
	     unsigned long length, unsigned long tail,
2112
	     struct buffer_page *tail_page, u64 ts)
S
Steven Rostedt 已提交
2113
{
2114
	struct buffer_page *commit_page = cpu_buffer->commit_page;
S
Steven Rostedt 已提交
2115
	struct ring_buffer *buffer = cpu_buffer->buffer;
S
Steven Rostedt 已提交
2116 2117
	struct buffer_page *next_page;
	int ret;
2118 2119 2120 2121 2122 2123 2124 2125 2126 2127 2128

	next_page = tail_page;

	rb_inc_page(cpu_buffer, &next_page);

	/*
	 * If for some reason, we had an interrupt storm that made
	 * it all the way around the buffer, bail, and warn
	 * about it.
	 */
	if (unlikely(next_page == commit_page)) {
S
Steven Rostedt 已提交
2129
		local_inc(&cpu_buffer->commit_overrun);
2130 2131 2132
		goto out_reset;
	}

S
Steven Rostedt 已提交
2133 2134 2135 2136 2137 2138 2139 2140 2141 2142 2143 2144 2145 2146 2147
	/*
	 * This is where the fun begins!
	 *
	 * We are fighting against races between a reader that
	 * could be on another CPU trying to swap its reader
	 * page with the buffer head.
	 *
	 * We are also fighting against interrupts coming in and
	 * moving the head or tail on us as well.
	 *
	 * If the next page is the head page then we have filled
	 * the buffer, unless the commit page is still on the
	 * reader page.
	 */
	if (rb_is_head_page(cpu_buffer, next_page, &tail_page->list)) {
2148

S
Steven Rostedt 已提交
2149 2150 2151 2152 2153 2154 2155 2156 2157 2158 2159 2160 2161 2162 2163 2164 2165 2166 2167 2168 2169 2170 2171 2172 2173 2174 2175 2176 2177 2178 2179 2180 2181 2182 2183 2184 2185
		/*
		 * If the commit is not on the reader page, then
		 * move the header page.
		 */
		if (!rb_is_reader_page(cpu_buffer->commit_page)) {
			/*
			 * If we are not in overwrite mode,
			 * this is easy, just stop here.
			 */
			if (!(buffer->flags & RB_FL_OVERWRITE))
				goto out_reset;

			ret = rb_handle_head_page(cpu_buffer,
						  tail_page,
						  next_page);
			if (ret < 0)
				goto out_reset;
			if (ret)
				goto out_again;
		} else {
			/*
			 * We need to be careful here too. The
			 * commit page could still be on the reader
			 * page. We could have a small buffer, and
			 * have filled up the buffer with events
			 * from interrupts and such, and wrapped.
			 *
			 * Note, if the tail page is also the on the
			 * reader_page, we let it move out.
			 */
			if (unlikely((cpu_buffer->commit_page !=
				      cpu_buffer->tail_page) &&
				     (cpu_buffer->commit_page ==
				      cpu_buffer->reader_page))) {
				local_inc(&cpu_buffer->commit_overrun);
				goto out_reset;
			}
2186 2187 2188
		}
	}

S
Steven Rostedt 已提交
2189 2190 2191 2192 2193 2194
	ret = rb_tail_page_update(cpu_buffer, tail_page, next_page);
	if (ret) {
		/*
		 * Nested commits always have zero deltas, so
		 * just reread the time stamp
		 */
2195 2196
		ts = rb_time_stamp(buffer);
		next_page->page->time_stamp = ts;
2197 2198
	}

S
Steven Rostedt 已提交
2199
 out_again:
2200

S
Steven Rostedt 已提交
2201
	rb_reset_tail(cpu_buffer, tail_page, tail, length);
2202 2203 2204 2205

	/* fail and let the caller try again */
	return ERR_PTR(-EAGAIN);

2206
 out_reset:
2207
	/* reset write */
2208
	rb_reset_tail(cpu_buffer, tail_page, tail, length);
2209

S
Steven Rostedt 已提交
2210
	return NULL;
S
Steven Rostedt 已提交
2211 2212
}

2213 2214
static struct ring_buffer_event *
__rb_reserve_next(struct ring_buffer_per_cpu *cpu_buffer,
2215 2216
		  unsigned long length, u64 ts,
		  u64 delta, int add_timestamp)
2217
{
2218
	struct buffer_page *tail_page;
2219 2220 2221
	struct ring_buffer_event *event;
	unsigned long tail, write;

2222 2223 2224 2225 2226 2227 2228 2229
	/*
	 * If the time delta since the last event is too big to
	 * hold in the time field of the event, then we append a
	 * TIME EXTEND event ahead of the data event.
	 */
	if (unlikely(add_timestamp))
		length += RB_LEN_TIME_EXTEND;

2230 2231
	tail_page = cpu_buffer->tail_page;
	write = local_add_return(length, &tail_page->write);
S
Steven Rostedt 已提交
2232 2233 2234

	/* set write to only the index of the write */
	write &= RB_WRITE_MASK;
2235 2236 2237
	tail = write - length;

	/* See if we shot pass the end of this buffer page */
2238
	if (unlikely(write > BUF_PAGE_SIZE))
2239
		return rb_move_tail(cpu_buffer, length, tail,
2240
				    tail_page, ts);
2241 2242 2243 2244

	/* We reserved something on the buffer */

	event = __rb_page_index(tail_page, tail);
2245
	kmemcheck_annotate_bitfield(event, bitfield);
2246
	rb_update_event(cpu_buffer, event, length, add_timestamp, delta);
2247

2248
	local_inc(&tail_page->entries);
2249 2250

	/*
2251 2252
	 * If this is the first commit on the page, then update
	 * its timestamp.
2253
	 */
2254
	if (!tail)
2255
		tail_page->page->time_stamp = ts;
2256

2257 2258 2259
	/* account for these added bytes */
	local_add(length, &cpu_buffer->entries_bytes);

2260 2261 2262
	return event;
}

2263 2264 2265 2266 2267 2268 2269 2270 2271 2272
static inline int
rb_try_to_discard(struct ring_buffer_per_cpu *cpu_buffer,
		  struct ring_buffer_event *event)
{
	unsigned long new_index, old_index;
	struct buffer_page *bpage;
	unsigned long index;
	unsigned long addr;

	new_index = rb_event_index(event);
2273
	old_index = new_index + rb_event_ts_length(event);
2274 2275 2276 2277 2278 2279
	addr = (unsigned long)event;
	addr &= PAGE_MASK;

	bpage = cpu_buffer->tail_page;

	if (bpage->page == (void *)addr && rb_page_write(bpage) == old_index) {
S
Steven Rostedt 已提交
2280 2281
		unsigned long write_mask =
			local_read(&bpage->write) & ~RB_WRITE_MASK;
2282
		unsigned long event_length = rb_event_length(event);
2283 2284 2285 2286 2287 2288
		/*
		 * This is on the tail page. It is possible that
		 * a write could come in and move the tail page
		 * and write to the next page. That is fine
		 * because we just shorten what is on this page.
		 */
S
Steven Rostedt 已提交
2289 2290
		old_index += write_mask;
		new_index += write_mask;
2291
		index = local_cmpxchg(&bpage->write, old_index, new_index);
2292 2293 2294
		if (index == old_index) {
			/* update counters */
			local_sub(event_length, &cpu_buffer->entries_bytes);
2295
			return 1;
2296
		}
2297 2298 2299 2300 2301 2302
	}

	/* could not discard */
	return 0;
}

2303 2304 2305 2306 2307 2308
static void rb_start_commit(struct ring_buffer_per_cpu *cpu_buffer)
{
	local_inc(&cpu_buffer->committing);
	local_inc(&cpu_buffer->commits);
}

2309
static inline void rb_end_commit(struct ring_buffer_per_cpu *cpu_buffer)
2310 2311 2312 2313 2314 2315 2316 2317 2318 2319 2320 2321 2322 2323 2324 2325 2326 2327 2328 2329 2330 2331 2332 2333 2334 2335 2336 2337 2338 2339 2340
{
	unsigned long commits;

	if (RB_WARN_ON(cpu_buffer,
		       !local_read(&cpu_buffer->committing)))
		return;

 again:
	commits = local_read(&cpu_buffer->commits);
	/* synchronize with interrupts */
	barrier();
	if (local_read(&cpu_buffer->committing) == 1)
		rb_set_commit_to_write(cpu_buffer);

	local_dec(&cpu_buffer->committing);

	/* synchronize with interrupts */
	barrier();

	/*
	 * Need to account for interrupts coming in between the
	 * updating of the commit page and the clearing of the
	 * committing counter.
	 */
	if (unlikely(local_read(&cpu_buffer->commits) != commits) &&
	    !local_read(&cpu_buffer->committing)) {
		local_inc(&cpu_buffer->committing);
		goto again;
	}
}

S
Steven Rostedt 已提交
2341
static struct ring_buffer_event *
2342 2343
rb_reserve_next_event(struct ring_buffer *buffer,
		      struct ring_buffer_per_cpu *cpu_buffer,
2344
		      unsigned long length)
S
Steven Rostedt 已提交
2345 2346
{
	struct ring_buffer_event *event;
2347
	u64 ts, delta;
2348
	int nr_loops = 0;
2349
	int add_timestamp;
2350
	u64 diff;
S
Steven Rostedt 已提交
2351

2352 2353
	rb_start_commit(cpu_buffer);

2354
#ifdef CONFIG_RING_BUFFER_ALLOW_SWAP
2355 2356 2357 2358 2359 2360 2361 2362 2363 2364 2365 2366
	/*
	 * Due to the ability to swap a cpu buffer from a buffer
	 * it is possible it was swapped before we committed.
	 * (committing stops a swap). We check for it here and
	 * if it happened, we have to fail the write.
	 */
	barrier();
	if (unlikely(ACCESS_ONCE(cpu_buffer->buffer) != buffer)) {
		local_dec(&cpu_buffer->committing);
		local_dec(&cpu_buffer->commits);
		return NULL;
	}
2367
#endif
2368

2369
	length = rb_calculate_event_length(length);
S
Steven Rostedt 已提交
2370
 again:
2371 2372 2373
	add_timestamp = 0;
	delta = 0;

2374 2375 2376 2377 2378 2379 2380 2381 2382
	/*
	 * We allow for interrupts to reenter here and do a trace.
	 * If one does, it will cause this original code to loop
	 * back here. Even with heavy interrupts happening, this
	 * should only happen a few times in a row. If this happens
	 * 1000 times in a row, there must be either an interrupt
	 * storm or we have something buggy.
	 * Bail!
	 */
S
Steven Rostedt 已提交
2383
	if (RB_WARN_ON(cpu_buffer, ++nr_loops > 1000))
2384
		goto out_fail;
2385

2386
	ts = rb_time_stamp(cpu_buffer->buffer);
2387
	diff = ts - cpu_buffer->write_stamp;
S
Steven Rostedt 已提交
2388

2389 2390
	/* make sure this diff is calculated here */
	barrier();
S
Steven Rostedt 已提交
2391

2392 2393
	/* Did the write stamp get updated already? */
	if (likely(ts >= cpu_buffer->write_stamp)) {
2394 2395
		delta = diff;
		if (unlikely(test_time_stamp(delta))) {
2396 2397 2398 2399
			int local_clock_stable = 1;
#ifdef CONFIG_HAVE_UNSTABLE_SCHED_CLOCK
			local_clock_stable = sched_clock_stable;
#endif
2400
			WARN_ONCE(delta > (1ULL << 59),
2401
				  KERN_WARNING "Delta way too big! %llu ts=%llu write stamp = %llu\n%s",
2402 2403
				  (unsigned long long)delta,
				  (unsigned long long)ts,
2404 2405 2406 2407 2408
				  (unsigned long long)cpu_buffer->write_stamp,
				  local_clock_stable ? "" :
				  "If you just came from a suspend/resume,\n"
				  "please switch to the trace global clock:\n"
				  "  echo global > /sys/kernel/debug/tracing/trace_clock\n");
2409
			add_timestamp = 1;
S
Steven Rostedt 已提交
2410
		}
2411
	}
S
Steven Rostedt 已提交
2412

2413 2414
	event = __rb_reserve_next(cpu_buffer, length, ts,
				  delta, add_timestamp);
2415
	if (unlikely(PTR_ERR(event) == -EAGAIN))
S
Steven Rostedt 已提交
2416 2417
		goto again;

2418 2419
	if (!event)
		goto out_fail;
S
Steven Rostedt 已提交
2420 2421

	return event;
2422 2423 2424 2425

 out_fail:
	rb_end_commit(cpu_buffer);
	return NULL;
S
Steven Rostedt 已提交
2426 2427
}

2428 2429
#ifdef CONFIG_TRACING

2430
#define TRACE_RECURSIVE_DEPTH 16
2431

2432 2433
/* Keep this code out of the fast path cache */
static noinline void trace_recursive_fail(void)
2434
{
2435 2436
	/* Disable all tracing before we do anything else */
	tracing_off_permanent();
2437

S
Steven Rostedt 已提交
2438
	printk_once(KERN_WARNING "Tracing recursion: depth[%ld]:"
2439
		    "HC[%lu]:SC[%lu]:NMI[%lu]\n",
2440
		    trace_recursion_buffer(),
2441 2442 2443
		    hardirq_count() >> HARDIRQ_SHIFT,
		    softirq_count() >> SOFTIRQ_SHIFT,
		    in_nmi());
2444

2445
	WARN_ON_ONCE(1);
2446 2447 2448 2449
}

static inline int trace_recursive_lock(void)
{
2450
	trace_recursion_inc();
2451

2452
	if (likely(trace_recursion_buffer() < TRACE_RECURSIVE_DEPTH))
2453 2454 2455 2456
		return 0;

	trace_recursive_fail();

2457
	return -1;
2458 2459
}

2460
static inline void trace_recursive_unlock(void)
2461
{
2462
	WARN_ON_ONCE(!trace_recursion_buffer());
2463

2464
	trace_recursion_dec();
2465 2466
}

2467 2468 2469 2470 2471 2472 2473
#else

#define trace_recursive_lock()		(0)
#define trace_recursive_unlock()	do { } while (0)

#endif

S
Steven Rostedt 已提交
2474 2475 2476 2477 2478 2479 2480 2481 2482 2483 2484 2485 2486 2487 2488 2489
/**
 * ring_buffer_lock_reserve - reserve a part of the buffer
 * @buffer: the ring buffer to reserve from
 * @length: the length of the data to reserve (excluding event header)
 *
 * Returns a reseverd event on the ring buffer to copy directly to.
 * The user of this interface will need to get the body to write into
 * and can use the ring_buffer_event_data() interface.
 *
 * The length is the length of the data needed, not the event length
 * which also includes the event header.
 *
 * Must be paired with ring_buffer_unlock_commit, unless NULL is returned.
 * If NULL is returned, then nothing has been allocated or locked.
 */
struct ring_buffer_event *
2490
ring_buffer_lock_reserve(struct ring_buffer *buffer, unsigned long length)
S
Steven Rostedt 已提交
2491 2492 2493
{
	struct ring_buffer_per_cpu *cpu_buffer;
	struct ring_buffer_event *event;
2494
	int cpu;
S
Steven Rostedt 已提交
2495

2496
	if (ring_buffer_flags != RB_BUFFERS_ON)
2497 2498
		return NULL;

S
Steven Rostedt 已提交
2499
	/* If we are tracing schedule, we don't want to recurse */
2500
	preempt_disable_notrace();
S
Steven Rostedt 已提交
2501

2502 2503 2504
	if (atomic_read(&buffer->record_disabled))
		goto out_nocheck;

2505 2506 2507
	if (trace_recursive_lock())
		goto out_nocheck;

S
Steven Rostedt 已提交
2508 2509
	cpu = raw_smp_processor_id();

2510
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
2511
		goto out;
S
Steven Rostedt 已提交
2512 2513 2514 2515

	cpu_buffer = buffer->buffers[cpu];

	if (atomic_read(&cpu_buffer->record_disabled))
2516
		goto out;
S
Steven Rostedt 已提交
2517

2518
	if (length > BUF_MAX_DATA_SIZE)
S
Steven Rostedt 已提交
2519
		goto out;
S
Steven Rostedt 已提交
2520

2521
	event = rb_reserve_next_event(buffer, cpu_buffer, length);
S
Steven Rostedt 已提交
2522
	if (!event)
2523
		goto out;
S
Steven Rostedt 已提交
2524 2525 2526

	return event;

2527
 out:
2528 2529 2530
	trace_recursive_unlock();

 out_nocheck:
2531
	preempt_enable_notrace();
S
Steven Rostedt 已提交
2532 2533
	return NULL;
}
2534
EXPORT_SYMBOL_GPL(ring_buffer_lock_reserve);
S
Steven Rostedt 已提交
2535

2536 2537
static void
rb_update_write_stamp(struct ring_buffer_per_cpu *cpu_buffer,
S
Steven Rostedt 已提交
2538 2539
		      struct ring_buffer_event *event)
{
2540 2541
	u64 delta;

2542 2543 2544 2545
	/*
	 * The event first in the commit queue updates the
	 * time stamp.
	 */
2546 2547 2548 2549 2550 2551 2552 2553 2554 2555 2556 2557 2558 2559 2560 2561
	if (rb_event_is_commit(cpu_buffer, event)) {
		/*
		 * A commit event that is first on a page
		 * updates the write timestamp with the page stamp
		 */
		if (!rb_event_index(event))
			cpu_buffer->write_stamp =
				cpu_buffer->commit_page->page->time_stamp;
		else if (event->type_len == RINGBUF_TYPE_TIME_EXTEND) {
			delta = event->array[0];
			delta <<= TS_SHIFT;
			delta += event->time_delta;
			cpu_buffer->write_stamp += delta;
		} else
			cpu_buffer->write_stamp += event->time_delta;
	}
2562
}
S
Steven Rostedt 已提交
2563

2564 2565 2566 2567 2568
static void rb_commit(struct ring_buffer_per_cpu *cpu_buffer,
		      struct ring_buffer_event *event)
{
	local_inc(&cpu_buffer->entries);
	rb_update_write_stamp(cpu_buffer, event);
2569
	rb_end_commit(cpu_buffer);
S
Steven Rostedt 已提交
2570 2571 2572 2573 2574 2575 2576 2577 2578 2579 2580 2581
}

/**
 * ring_buffer_unlock_commit - commit a reserved
 * @buffer: The buffer to commit to
 * @event: The event pointer to commit.
 *
 * This commits the data to the ring buffer, and releases any locks held.
 *
 * Must be paired with ring_buffer_lock_reserve.
 */
int ring_buffer_unlock_commit(struct ring_buffer *buffer,
2582
			      struct ring_buffer_event *event)
S
Steven Rostedt 已提交
2583 2584 2585 2586 2587 2588 2589 2590
{
	struct ring_buffer_per_cpu *cpu_buffer;
	int cpu = raw_smp_processor_id();

	cpu_buffer = buffer->buffers[cpu];

	rb_commit(cpu_buffer, event);

2591 2592
	trace_recursive_unlock();

2593
	preempt_enable_notrace();
S
Steven Rostedt 已提交
2594 2595 2596

	return 0;
}
2597
EXPORT_SYMBOL_GPL(ring_buffer_unlock_commit);
S
Steven Rostedt 已提交
2598

2599 2600
static inline void rb_event_discard(struct ring_buffer_event *event)
{
2601 2602 2603
	if (event->type_len == RINGBUF_TYPE_TIME_EXTEND)
		event = skip_time_extend(event);

2604 2605 2606
	/* array[0] holds the actual length for the discarded event */
	event->array[0] = rb_event_data_length(event) - RB_EVNT_HDR_SIZE;
	event->type_len = RINGBUF_TYPE_PADDING;
2607 2608 2609 2610 2611
	/* time delta must be non zero */
	if (!event->time_delta)
		event->time_delta = 1;
}

2612 2613 2614 2615 2616 2617 2618 2619 2620 2621 2622 2623 2624 2625 2626 2627 2628 2629 2630 2631 2632 2633 2634 2635 2636 2637 2638 2639 2640 2641 2642 2643 2644 2645 2646 2647 2648 2649 2650 2651
/*
 * Decrement the entries to the page that an event is on.
 * The event does not even need to exist, only the pointer
 * to the page it is on. This may only be called before the commit
 * takes place.
 */
static inline void
rb_decrement_entry(struct ring_buffer_per_cpu *cpu_buffer,
		   struct ring_buffer_event *event)
{
	unsigned long addr = (unsigned long)event;
	struct buffer_page *bpage = cpu_buffer->commit_page;
	struct buffer_page *start;

	addr &= PAGE_MASK;

	/* Do the likely case first */
	if (likely(bpage->page == (void *)addr)) {
		local_dec(&bpage->entries);
		return;
	}

	/*
	 * Because the commit page may be on the reader page we
	 * start with the next page and check the end loop there.
	 */
	rb_inc_page(cpu_buffer, &bpage);
	start = bpage;
	do {
		if (bpage->page == (void *)addr) {
			local_dec(&bpage->entries);
			return;
		}
		rb_inc_page(cpu_buffer, &bpage);
	} while (bpage != start);

	/* commit not part of this buffer?? */
	RB_WARN_ON(cpu_buffer, 1);
}

2652 2653 2654 2655 2656
/**
 * ring_buffer_commit_discard - discard an event that has not been committed
 * @buffer: the ring buffer
 * @event: non committed event to discard
 *
2657 2658 2659 2660 2661 2662
 * Sometimes an event that is in the ring buffer needs to be ignored.
 * This function lets the user discard an event in the ring buffer
 * and then that event will not be read later.
 *
 * This function only works if it is called before the the item has been
 * committed. It will try to free the event from the ring buffer
2663 2664 2665 2666 2667 2668 2669 2670 2671 2672 2673 2674 2675 2676 2677
 * if another event has not been added behind it.
 *
 * If another event has been added behind it, it will set the event
 * up as discarded, and perform the commit.
 *
 * If this function is called, do not call ring_buffer_unlock_commit on
 * the event.
 */
void ring_buffer_discard_commit(struct ring_buffer *buffer,
				struct ring_buffer_event *event)
{
	struct ring_buffer_per_cpu *cpu_buffer;
	int cpu;

	/* The event is discarded regardless */
2678
	rb_event_discard(event);
2679

2680 2681 2682
	cpu = smp_processor_id();
	cpu_buffer = buffer->buffers[cpu];

2683 2684 2685 2686 2687
	/*
	 * This must only be called if the event has not been
	 * committed yet. Thus we can assume that preemption
	 * is still disabled.
	 */
2688
	RB_WARN_ON(buffer, !local_read(&cpu_buffer->committing));
2689

2690
	rb_decrement_entry(cpu_buffer, event);
2691
	if (rb_try_to_discard(cpu_buffer, event))
2692
		goto out;
2693 2694 2695

	/*
	 * The commit is still visible by the reader, so we
2696
	 * must still update the timestamp.
2697
	 */
2698
	rb_update_write_stamp(cpu_buffer, event);
2699
 out:
2700
	rb_end_commit(cpu_buffer);
2701

2702 2703
	trace_recursive_unlock();

2704
	preempt_enable_notrace();
2705 2706 2707 2708

}
EXPORT_SYMBOL_GPL(ring_buffer_discard_commit);

S
Steven Rostedt 已提交
2709 2710 2711 2712 2713 2714 2715 2716 2717 2718 2719 2720 2721 2722 2723 2724 2725 2726 2727 2728 2729
/**
 * ring_buffer_write - write data to the buffer without reserving
 * @buffer: The ring buffer to write to.
 * @length: The length of the data being written (excluding the event header)
 * @data: The data to write to the buffer.
 *
 * This is like ring_buffer_lock_reserve and ring_buffer_unlock_commit as
 * one function. If you already have the data to write to the buffer, it
 * may be easier to simply call this function.
 *
 * Note, like ring_buffer_lock_reserve, the length is the length of the data
 * and not the length of the event which would hold the header.
 */
int ring_buffer_write(struct ring_buffer *buffer,
			unsigned long length,
			void *data)
{
	struct ring_buffer_per_cpu *cpu_buffer;
	struct ring_buffer_event *event;
	void *body;
	int ret = -EBUSY;
2730
	int cpu;
S
Steven Rostedt 已提交
2731

2732
	if (ring_buffer_flags != RB_BUFFERS_ON)
2733 2734
		return -EBUSY;

2735
	preempt_disable_notrace();
S
Steven Rostedt 已提交
2736

2737 2738 2739
	if (atomic_read(&buffer->record_disabled))
		goto out;

S
Steven Rostedt 已提交
2740 2741
	cpu = raw_smp_processor_id();

2742
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
2743
		goto out;
S
Steven Rostedt 已提交
2744 2745 2746 2747 2748 2749

	cpu_buffer = buffer->buffers[cpu];

	if (atomic_read(&cpu_buffer->record_disabled))
		goto out;

2750 2751 2752
	if (length > BUF_MAX_DATA_SIZE)
		goto out;

2753
	event = rb_reserve_next_event(buffer, cpu_buffer, length);
S
Steven Rostedt 已提交
2754 2755 2756 2757 2758 2759 2760 2761 2762 2763 2764
	if (!event)
		goto out;

	body = rb_event_data(event);

	memcpy(body, data, length);

	rb_commit(cpu_buffer, event);

	ret = 0;
 out:
2765
	preempt_enable_notrace();
S
Steven Rostedt 已提交
2766 2767 2768

	return ret;
}
2769
EXPORT_SYMBOL_GPL(ring_buffer_write);
S
Steven Rostedt 已提交
2770

2771
static int rb_per_cpu_empty(struct ring_buffer_per_cpu *cpu_buffer)
S
Steven Rostedt 已提交
2772 2773
{
	struct buffer_page *reader = cpu_buffer->reader_page;
S
Steven Rostedt 已提交
2774
	struct buffer_page *head = rb_set_head_page(cpu_buffer);
S
Steven Rostedt 已提交
2775 2776
	struct buffer_page *commit = cpu_buffer->commit_page;

S
Steven Rostedt 已提交
2777 2778 2779 2780
	/* In case of error, head will be NULL */
	if (unlikely(!head))
		return 1;

S
Steven Rostedt 已提交
2781 2782 2783 2784 2785 2786
	return reader->read == rb_page_commit(reader) &&
		(commit == reader ||
		 (commit == head &&
		  head->read == rb_page_commit(commit)));
}

S
Steven Rostedt 已提交
2787 2788 2789 2790 2791 2792 2793 2794 2795 2796 2797 2798 2799
/**
 * ring_buffer_record_disable - stop all writes into the buffer
 * @buffer: The ring buffer to stop writes to.
 *
 * This prevents all writes to the buffer. Any attempt to write
 * to the buffer after this will fail and return NULL.
 *
 * The caller should call synchronize_sched() after this.
 */
void ring_buffer_record_disable(struct ring_buffer *buffer)
{
	atomic_inc(&buffer->record_disabled);
}
2800
EXPORT_SYMBOL_GPL(ring_buffer_record_disable);
S
Steven Rostedt 已提交
2801 2802 2803 2804 2805 2806

/**
 * ring_buffer_record_enable - enable writes to the buffer
 * @buffer: The ring buffer to enable writes
 *
 * Note, multiple disables will need the same number of enables
2807
 * to truly enable the writing (much like preempt_disable).
S
Steven Rostedt 已提交
2808 2809 2810 2811 2812
 */
void ring_buffer_record_enable(struct ring_buffer *buffer)
{
	atomic_dec(&buffer->record_disabled);
}
2813
EXPORT_SYMBOL_GPL(ring_buffer_record_enable);
S
Steven Rostedt 已提交
2814

2815 2816 2817 2818 2819 2820 2821 2822
/**
 * ring_buffer_record_off - stop all writes into the buffer
 * @buffer: The ring buffer to stop writes to.
 *
 * This prevents all writes to the buffer. Any attempt to write
 * to the buffer after this will fail and return NULL.
 *
 * This is different than ring_buffer_record_disable() as
2823
 * it works like an on/off switch, where as the disable() version
2824 2825 2826 2827 2828 2829 2830 2831 2832 2833 2834 2835 2836 2837 2838 2839 2840 2841 2842 2843 2844 2845
 * must be paired with a enable().
 */
void ring_buffer_record_off(struct ring_buffer *buffer)
{
	unsigned int rd;
	unsigned int new_rd;

	do {
		rd = atomic_read(&buffer->record_disabled);
		new_rd = rd | RB_BUFFER_OFF;
	} while (atomic_cmpxchg(&buffer->record_disabled, rd, new_rd) != rd);
}
EXPORT_SYMBOL_GPL(ring_buffer_record_off);

/**
 * ring_buffer_record_on - restart writes into the buffer
 * @buffer: The ring buffer to start writes to.
 *
 * This enables all writes to the buffer that was disabled by
 * ring_buffer_record_off().
 *
 * This is different than ring_buffer_record_enable() as
2846
 * it works like an on/off switch, where as the enable() version
2847 2848 2849 2850 2851 2852 2853 2854 2855 2856 2857 2858 2859 2860 2861 2862 2863 2864 2865 2866 2867 2868 2869 2870 2871
 * must be paired with a disable().
 */
void ring_buffer_record_on(struct ring_buffer *buffer)
{
	unsigned int rd;
	unsigned int new_rd;

	do {
		rd = atomic_read(&buffer->record_disabled);
		new_rd = rd & ~RB_BUFFER_OFF;
	} while (atomic_cmpxchg(&buffer->record_disabled, rd, new_rd) != rd);
}
EXPORT_SYMBOL_GPL(ring_buffer_record_on);

/**
 * ring_buffer_record_is_on - return true if the ring buffer can write
 * @buffer: The ring buffer to see if write is enabled
 *
 * Returns true if the ring buffer is in a state that it accepts writes.
 */
int ring_buffer_record_is_on(struct ring_buffer *buffer)
{
	return !atomic_read(&buffer->record_disabled);
}

S
Steven Rostedt 已提交
2872 2873 2874 2875 2876 2877 2878 2879 2880 2881 2882 2883 2884 2885
/**
 * ring_buffer_record_disable_cpu - stop all writes into the cpu_buffer
 * @buffer: The ring buffer to stop writes to.
 * @cpu: The CPU buffer to stop
 *
 * This prevents all writes to the buffer. Any attempt to write
 * to the buffer after this will fail and return NULL.
 *
 * The caller should call synchronize_sched() after this.
 */
void ring_buffer_record_disable_cpu(struct ring_buffer *buffer, int cpu)
{
	struct ring_buffer_per_cpu *cpu_buffer;

2886
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
2887
		return;
S
Steven Rostedt 已提交
2888 2889 2890 2891

	cpu_buffer = buffer->buffers[cpu];
	atomic_inc(&cpu_buffer->record_disabled);
}
2892
EXPORT_SYMBOL_GPL(ring_buffer_record_disable_cpu);
S
Steven Rostedt 已提交
2893 2894 2895 2896 2897 2898 2899

/**
 * ring_buffer_record_enable_cpu - enable writes to the buffer
 * @buffer: The ring buffer to enable writes
 * @cpu: The CPU to enable.
 *
 * Note, multiple disables will need the same number of enables
2900
 * to truly enable the writing (much like preempt_disable).
S
Steven Rostedt 已提交
2901 2902 2903 2904 2905
 */
void ring_buffer_record_enable_cpu(struct ring_buffer *buffer, int cpu)
{
	struct ring_buffer_per_cpu *cpu_buffer;

2906
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
2907
		return;
S
Steven Rostedt 已提交
2908 2909 2910 2911

	cpu_buffer = buffer->buffers[cpu];
	atomic_dec(&cpu_buffer->record_disabled);
}
2912
EXPORT_SYMBOL_GPL(ring_buffer_record_enable_cpu);
S
Steven Rostedt 已提交
2913

2914 2915 2916 2917 2918 2919 2920 2921 2922 2923 2924 2925 2926
/*
 * The total entries in the ring buffer is the running counter
 * of entries entered into the ring buffer, minus the sum of
 * the entries read from the ring buffer and the number of
 * entries that were overwritten.
 */
static inline unsigned long
rb_num_of_entries(struct ring_buffer_per_cpu *cpu_buffer)
{
	return local_read(&cpu_buffer->entries) -
		(local_read(&cpu_buffer->overrun) + cpu_buffer->read);
}

2927 2928 2929 2930 2931 2932 2933 2934 2935 2936 2937 2938 2939 2940 2941 2942
/**
 * ring_buffer_oldest_event_ts - get the oldest event timestamp from the buffer
 * @buffer: The ring buffer
 * @cpu: The per CPU buffer to read from.
 */
unsigned long ring_buffer_oldest_event_ts(struct ring_buffer *buffer, int cpu)
{
	unsigned long flags;
	struct ring_buffer_per_cpu *cpu_buffer;
	struct buffer_page *bpage;
	unsigned long ret;

	if (!cpumask_test_cpu(cpu, buffer->cpumask))
		return 0;

	cpu_buffer = buffer->buffers[cpu];
2943
	raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
2944 2945 2946 2947 2948 2949 2950 2951 2952
	/*
	 * if the tail is on reader_page, oldest time stamp is on the reader
	 * page
	 */
	if (cpu_buffer->tail_page == cpu_buffer->reader_page)
		bpage = cpu_buffer->reader_page;
	else
		bpage = rb_set_head_page(cpu_buffer);
	ret = bpage->page->time_stamp;
2953
	raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
2954 2955 2956 2957 2958 2959 2960 2961 2962 2963 2964 2965 2966 2967 2968 2969 2970 2971 2972 2973 2974 2975 2976 2977 2978

	return ret;
}
EXPORT_SYMBOL_GPL(ring_buffer_oldest_event_ts);

/**
 * ring_buffer_bytes_cpu - get the number of bytes consumed in a cpu buffer
 * @buffer: The ring buffer
 * @cpu: The per CPU buffer to read from.
 */
unsigned long ring_buffer_bytes_cpu(struct ring_buffer *buffer, int cpu)
{
	struct ring_buffer_per_cpu *cpu_buffer;
	unsigned long ret;

	if (!cpumask_test_cpu(cpu, buffer->cpumask))
		return 0;

	cpu_buffer = buffer->buffers[cpu];
	ret = local_read(&cpu_buffer->entries_bytes) - cpu_buffer->read_bytes;

	return ret;
}
EXPORT_SYMBOL_GPL(ring_buffer_bytes_cpu);

S
Steven Rostedt 已提交
2979 2980 2981 2982 2983 2984 2985 2986 2987
/**
 * ring_buffer_entries_cpu - get the number of entries in a cpu buffer
 * @buffer: The ring buffer
 * @cpu: The per CPU buffer to get the entries from.
 */
unsigned long ring_buffer_entries_cpu(struct ring_buffer *buffer, int cpu)
{
	struct ring_buffer_per_cpu *cpu_buffer;

2988
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
2989
		return 0;
S
Steven Rostedt 已提交
2990 2991

	cpu_buffer = buffer->buffers[cpu];
2992

2993
	return rb_num_of_entries(cpu_buffer);
S
Steven Rostedt 已提交
2994
}
2995
EXPORT_SYMBOL_GPL(ring_buffer_entries_cpu);
S
Steven Rostedt 已提交
2996 2997 2998 2999 3000 3001 3002 3003 3004

/**
 * ring_buffer_overrun_cpu - get the number of overruns in a cpu_buffer
 * @buffer: The ring buffer
 * @cpu: The per CPU buffer to get the number of overruns from
 */
unsigned long ring_buffer_overrun_cpu(struct ring_buffer *buffer, int cpu)
{
	struct ring_buffer_per_cpu *cpu_buffer;
3005
	unsigned long ret;
S
Steven Rostedt 已提交
3006

3007
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
3008
		return 0;
S
Steven Rostedt 已提交
3009 3010

	cpu_buffer = buffer->buffers[cpu];
S
Steven Rostedt 已提交
3011
	ret = local_read(&cpu_buffer->overrun);
3012 3013

	return ret;
S
Steven Rostedt 已提交
3014
}
3015
EXPORT_SYMBOL_GPL(ring_buffer_overrun_cpu);
S
Steven Rostedt 已提交
3016

3017 3018 3019 3020 3021 3022 3023 3024 3025 3026 3027 3028 3029 3030 3031
/**
 * ring_buffer_commit_overrun_cpu - get the number of overruns caused by commits
 * @buffer: The ring buffer
 * @cpu: The per CPU buffer to get the number of overruns from
 */
unsigned long
ring_buffer_commit_overrun_cpu(struct ring_buffer *buffer, int cpu)
{
	struct ring_buffer_per_cpu *cpu_buffer;
	unsigned long ret;

	if (!cpumask_test_cpu(cpu, buffer->cpumask))
		return 0;

	cpu_buffer = buffer->buffers[cpu];
S
Steven Rostedt 已提交
3032
	ret = local_read(&cpu_buffer->commit_overrun);
3033 3034 3035 3036 3037

	return ret;
}
EXPORT_SYMBOL_GPL(ring_buffer_commit_overrun_cpu);

S
Steven Rostedt 已提交
3038 3039 3040 3041 3042 3043 3044 3045 3046 3047 3048 3049 3050 3051 3052 3053
/**
 * ring_buffer_entries - get the number of entries in a buffer
 * @buffer: The ring buffer
 *
 * Returns the total number of entries in the ring buffer
 * (all CPU entries)
 */
unsigned long ring_buffer_entries(struct ring_buffer *buffer)
{
	struct ring_buffer_per_cpu *cpu_buffer;
	unsigned long entries = 0;
	int cpu;

	/* if you care about this being correct, lock the buffer */
	for_each_buffer_cpu(buffer, cpu) {
		cpu_buffer = buffer->buffers[cpu];
3054
		entries += rb_num_of_entries(cpu_buffer);
S
Steven Rostedt 已提交
3055 3056 3057 3058
	}

	return entries;
}
3059
EXPORT_SYMBOL_GPL(ring_buffer_entries);
S
Steven Rostedt 已提交
3060 3061

/**
3062
 * ring_buffer_overruns - get the number of overruns in buffer
S
Steven Rostedt 已提交
3063 3064 3065 3066 3067 3068 3069 3070 3071 3072 3073 3074 3075 3076
 * @buffer: The ring buffer
 *
 * Returns the total number of overruns in the ring buffer
 * (all CPU entries)
 */
unsigned long ring_buffer_overruns(struct ring_buffer *buffer)
{
	struct ring_buffer_per_cpu *cpu_buffer;
	unsigned long overruns = 0;
	int cpu;

	/* if you care about this being correct, lock the buffer */
	for_each_buffer_cpu(buffer, cpu) {
		cpu_buffer = buffer->buffers[cpu];
S
Steven Rostedt 已提交
3077
		overruns += local_read(&cpu_buffer->overrun);
S
Steven Rostedt 已提交
3078 3079 3080 3081
	}

	return overruns;
}
3082
EXPORT_SYMBOL_GPL(ring_buffer_overruns);
S
Steven Rostedt 已提交
3083

3084
static void rb_iter_reset(struct ring_buffer_iter *iter)
S
Steven Rostedt 已提交
3085 3086 3087
{
	struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;

3088 3089
	/* Iterator usage is expected to have record disabled */
	if (list_empty(&cpu_buffer->reader_page->list)) {
S
Steven Rostedt 已提交
3090 3091 3092 3093
		iter->head_page = rb_set_head_page(cpu_buffer);
		if (unlikely(!iter->head_page))
			return;
		iter->head = iter->head_page->read;
3094 3095
	} else {
		iter->head_page = cpu_buffer->reader_page;
3096
		iter->head = cpu_buffer->reader_page->read;
3097 3098 3099 3100
	}
	if (iter->head)
		iter->read_stamp = cpu_buffer->read_stamp;
	else
3101
		iter->read_stamp = iter->head_page->page->time_stamp;
3102 3103
	iter->cache_reader_page = cpu_buffer->reader_page;
	iter->cache_read = cpu_buffer->read;
3104
}
S
Steven Rostedt 已提交
3105

3106 3107 3108 3109 3110 3111 3112 3113 3114
/**
 * ring_buffer_iter_reset - reset an iterator
 * @iter: The iterator to reset
 *
 * Resets the iterator, so that it will start from the beginning
 * again.
 */
void ring_buffer_iter_reset(struct ring_buffer_iter *iter)
{
3115
	struct ring_buffer_per_cpu *cpu_buffer;
3116 3117
	unsigned long flags;

3118 3119 3120 3121 3122
	if (!iter)
		return;

	cpu_buffer = iter->cpu_buffer;

3123
	raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
3124
	rb_iter_reset(iter);
3125
	raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
S
Steven Rostedt 已提交
3126
}
3127
EXPORT_SYMBOL_GPL(ring_buffer_iter_reset);
S
Steven Rostedt 已提交
3128 3129 3130 3131 3132 3133 3134 3135 3136 3137 3138

/**
 * ring_buffer_iter_empty - check if an iterator has no more to read
 * @iter: The iterator to check
 */
int ring_buffer_iter_empty(struct ring_buffer_iter *iter)
{
	struct ring_buffer_per_cpu *cpu_buffer;

	cpu_buffer = iter->cpu_buffer;

S
Steven Rostedt 已提交
3139 3140
	return iter->head_page == cpu_buffer->commit_page &&
		iter->head == rb_commit_index(cpu_buffer);
S
Steven Rostedt 已提交
3141
}
3142
EXPORT_SYMBOL_GPL(ring_buffer_iter_empty);
S
Steven Rostedt 已提交
3143 3144 3145 3146 3147 3148 3149

static void
rb_update_read_stamp(struct ring_buffer_per_cpu *cpu_buffer,
		     struct ring_buffer_event *event)
{
	u64 delta;

3150
	switch (event->type_len) {
S
Steven Rostedt 已提交
3151 3152 3153 3154 3155 3156 3157 3158 3159 3160 3161 3162 3163 3164 3165 3166 3167 3168 3169 3170 3171 3172 3173 3174 3175 3176 3177 3178 3179 3180
	case RINGBUF_TYPE_PADDING:
		return;

	case RINGBUF_TYPE_TIME_EXTEND:
		delta = event->array[0];
		delta <<= TS_SHIFT;
		delta += event->time_delta;
		cpu_buffer->read_stamp += delta;
		return;

	case RINGBUF_TYPE_TIME_STAMP:
		/* FIXME: not implemented */
		return;

	case RINGBUF_TYPE_DATA:
		cpu_buffer->read_stamp += event->time_delta;
		return;

	default:
		BUG();
	}
	return;
}

static void
rb_update_iter_read_stamp(struct ring_buffer_iter *iter,
			  struct ring_buffer_event *event)
{
	u64 delta;

3181
	switch (event->type_len) {
S
Steven Rostedt 已提交
3182 3183 3184 3185 3186 3187 3188 3189 3190 3191 3192 3193 3194 3195 3196 3197 3198 3199 3200 3201 3202 3203 3204 3205
	case RINGBUF_TYPE_PADDING:
		return;

	case RINGBUF_TYPE_TIME_EXTEND:
		delta = event->array[0];
		delta <<= TS_SHIFT;
		delta += event->time_delta;
		iter->read_stamp += delta;
		return;

	case RINGBUF_TYPE_TIME_STAMP:
		/* FIXME: not implemented */
		return;

	case RINGBUF_TYPE_DATA:
		iter->read_stamp += event->time_delta;
		return;

	default:
		BUG();
	}
	return;
}

3206 3207
static struct buffer_page *
rb_get_reader_page(struct ring_buffer_per_cpu *cpu_buffer)
S
Steven Rostedt 已提交
3208
{
3209
	struct buffer_page *reader = NULL;
3210
	unsigned long overwrite;
3211
	unsigned long flags;
3212
	int nr_loops = 0;
S
Steven Rostedt 已提交
3213
	int ret;
3214

3215
	local_irq_save(flags);
3216
	arch_spin_lock(&cpu_buffer->lock);
3217 3218

 again:
3219 3220 3221 3222 3223 3224
	/*
	 * This should normally only loop twice. But because the
	 * start of the reader inserts an empty page, it causes
	 * a case where we will loop three times. There should be no
	 * reason to loop four times (that I know of).
	 */
S
Steven Rostedt 已提交
3225
	if (RB_WARN_ON(cpu_buffer, ++nr_loops > 3)) {
3226 3227 3228 3229
		reader = NULL;
		goto out;
	}

3230 3231 3232
	reader = cpu_buffer->reader_page;

	/* If there's more to read, return this page */
S
Steven Rostedt 已提交
3233
	if (cpu_buffer->reader_page->read < rb_page_size(reader))
3234 3235 3236
		goto out;

	/* Never should we have an index greater than the size */
S
Steven Rostedt 已提交
3237 3238 3239
	if (RB_WARN_ON(cpu_buffer,
		       cpu_buffer->reader_page->read > rb_page_size(reader)))
		goto out;
3240 3241 3242

	/* check if we caught up to the tail */
	reader = NULL;
S
Steven Rostedt 已提交
3243
	if (cpu_buffer->commit_page == cpu_buffer->reader_page)
3244
		goto out;
S
Steven Rostedt 已提交
3245

3246 3247 3248 3249
	/* Don't bother swapping if the ring buffer is empty */
	if (rb_num_of_entries(cpu_buffer) == 0)
		goto out;

S
Steven Rostedt 已提交
3250
	/*
3251
	 * Reset the reader page to size zero.
S
Steven Rostedt 已提交
3252
	 */
S
Steven Rostedt 已提交
3253 3254 3255
	local_set(&cpu_buffer->reader_page->write, 0);
	local_set(&cpu_buffer->reader_page->entries, 0);
	local_set(&cpu_buffer->reader_page->page->commit, 0);
3256
	cpu_buffer->reader_page->real_end = 0;
S
Steven Rostedt 已提交
3257

S
Steven Rostedt 已提交
3258 3259 3260 3261 3262
 spin:
	/*
	 * Splice the empty reader page into the list around the head.
	 */
	reader = rb_set_head_page(cpu_buffer);
3263
	cpu_buffer->reader_page->list.next = rb_list_head(reader->list.next);
3264
	cpu_buffer->reader_page->list.prev = reader->list.prev;
S
Steven Rostedt 已提交
3265

3266 3267 3268
	/*
	 * cpu_buffer->pages just needs to point to the buffer, it
	 *  has no specific buffer page to point to. Lets move it out
L
Lucas De Marchi 已提交
3269
	 *  of our way so we don't accidentally swap it.
3270 3271 3272
	 */
	cpu_buffer->pages = reader->list.prev;

S
Steven Rostedt 已提交
3273 3274
	/* The reader page will be pointing to the new head */
	rb_set_list_to_head(cpu_buffer, &cpu_buffer->reader_page->list);
S
Steven Rostedt 已提交
3275

3276 3277 3278 3279 3280 3281 3282 3283 3284 3285 3286 3287
	/*
	 * We want to make sure we read the overruns after we set up our
	 * pointers to the next object. The writer side does a
	 * cmpxchg to cross pages which acts as the mb on the writer
	 * side. Note, the reader will constantly fail the swap
	 * while the writer is updating the pointers, so this
	 * guarantees that the overwrite recorded here is the one we
	 * want to compare with the last_overrun.
	 */
	smp_mb();
	overwrite = local_read(&(cpu_buffer->overrun));

S
Steven Rostedt 已提交
3288 3289 3290 3291 3292 3293 3294 3295 3296 3297 3298 3299
	/*
	 * Here's the tricky part.
	 *
	 * We need to move the pointer past the header page.
	 * But we can only do that if a writer is not currently
	 * moving it. The page before the header page has the
	 * flag bit '1' set if it is pointing to the page we want.
	 * but if the writer is in the process of moving it
	 * than it will be '2' or already moved '0'.
	 */

	ret = rb_head_page_replace(reader, cpu_buffer->reader_page);
S
Steven Rostedt 已提交
3300 3301

	/*
S
Steven Rostedt 已提交
3302
	 * If we did not convert it, then we must try again.
S
Steven Rostedt 已提交
3303
	 */
S
Steven Rostedt 已提交
3304 3305
	if (!ret)
		goto spin;
S
Steven Rostedt 已提交
3306

S
Steven Rostedt 已提交
3307 3308 3309 3310 3311
	/*
	 * Yeah! We succeeded in replacing the page.
	 *
	 * Now make the new head point back to the reader page.
	 */
3312
	rb_list_head(reader->list.next)->prev = &cpu_buffer->reader_page->list;
S
Steven Rostedt 已提交
3313
	rb_inc_page(cpu_buffer, &cpu_buffer->head_page);
3314 3315 3316 3317 3318

	/* Finally update the reader page to the new head */
	cpu_buffer->reader_page = reader;
	rb_reset_reader_page(cpu_buffer);

3319 3320 3321 3322 3323
	if (overwrite != cpu_buffer->last_overrun) {
		cpu_buffer->lost_events = overwrite - cpu_buffer->last_overrun;
		cpu_buffer->last_overrun = overwrite;
	}

3324 3325 3326
	goto again;

 out:
3327
	arch_spin_unlock(&cpu_buffer->lock);
3328
	local_irq_restore(flags);
3329 3330 3331 3332 3333 3334 3335 3336 3337 3338 3339

	return reader;
}

static void rb_advance_reader(struct ring_buffer_per_cpu *cpu_buffer)
{
	struct ring_buffer_event *event;
	struct buffer_page *reader;
	unsigned length;

	reader = rb_get_reader_page(cpu_buffer);
S
Steven Rostedt 已提交
3340

3341
	/* This function should not be called when buffer is empty */
S
Steven Rostedt 已提交
3342 3343
	if (RB_WARN_ON(cpu_buffer, !reader))
		return;
S
Steven Rostedt 已提交
3344

3345 3346
	event = rb_reader_event(cpu_buffer);

3347
	if (event->type_len <= RINGBUF_TYPE_DATA_TYPE_LEN_MAX)
3348
		cpu_buffer->read++;
3349 3350 3351 3352

	rb_update_read_stamp(cpu_buffer, event);

	length = rb_event_length(event);
3353
	cpu_buffer->reader_page->read += length;
S
Steven Rostedt 已提交
3354 3355 3356 3357 3358 3359 3360 3361 3362 3363 3364 3365 3366
}

static void rb_advance_iter(struct ring_buffer_iter *iter)
{
	struct ring_buffer_per_cpu *cpu_buffer;
	struct ring_buffer_event *event;
	unsigned length;

	cpu_buffer = iter->cpu_buffer;

	/*
	 * Check if we are at the end of the buffer.
	 */
S
Steven Rostedt 已提交
3367
	if (iter->head >= rb_page_size(iter->head_page)) {
3368 3369
		/* discarded commits can make the page empty */
		if (iter->head_page == cpu_buffer->commit_page)
S
Steven Rostedt 已提交
3370
			return;
3371
		rb_inc_iter(iter);
S
Steven Rostedt 已提交
3372 3373 3374 3375 3376 3377 3378 3379 3380 3381 3382
		return;
	}

	event = rb_iter_head_event(iter);

	length = rb_event_length(event);

	/*
	 * This should not be called to advance the header if we are
	 * at the tail of the buffer.
	 */
S
Steven Rostedt 已提交
3383
	if (RB_WARN_ON(cpu_buffer,
3384
		       (iter->head_page == cpu_buffer->commit_page) &&
S
Steven Rostedt 已提交
3385 3386
		       (iter->head + length > rb_commit_index(cpu_buffer))))
		return;
S
Steven Rostedt 已提交
3387 3388 3389 3390 3391 3392

	rb_update_iter_read_stamp(iter, event);

	iter->head += length;

	/* check for end of page padding */
S
Steven Rostedt 已提交
3393 3394
	if ((iter->head >= rb_page_size(iter->head_page)) &&
	    (iter->head_page != cpu_buffer->commit_page))
S
Steven Rostedt 已提交
3395 3396 3397
		rb_advance_iter(iter);
}

3398 3399 3400 3401 3402
static int rb_lost_events(struct ring_buffer_per_cpu *cpu_buffer)
{
	return cpu_buffer->lost_events;
}

S
Steven Rostedt 已提交
3403
static struct ring_buffer_event *
3404 3405
rb_buffer_peek(struct ring_buffer_per_cpu *cpu_buffer, u64 *ts,
	       unsigned long *lost_events)
S
Steven Rostedt 已提交
3406 3407
{
	struct ring_buffer_event *event;
3408
	struct buffer_page *reader;
3409
	int nr_loops = 0;
S
Steven Rostedt 已提交
3410 3411

 again:
3412
	/*
3413 3414 3415 3416
	 * We repeat when a time extend is encountered.
	 * Since the time extend is always attached to a data event,
	 * we should never loop more than once.
	 * (We never hit the following condition more than twice).
3417
	 */
3418
	if (RB_WARN_ON(cpu_buffer, ++nr_loops > 2))
3419 3420
		return NULL;

3421 3422
	reader = rb_get_reader_page(cpu_buffer);
	if (!reader)
S
Steven Rostedt 已提交
3423 3424
		return NULL;

3425
	event = rb_reader_event(cpu_buffer);
S
Steven Rostedt 已提交
3426

3427
	switch (event->type_len) {
S
Steven Rostedt 已提交
3428
	case RINGBUF_TYPE_PADDING:
3429 3430 3431 3432 3433 3434 3435 3436 3437 3438 3439
		if (rb_null_event(event))
			RB_WARN_ON(cpu_buffer, 1);
		/*
		 * Because the writer could be discarding every
		 * event it creates (which would probably be bad)
		 * if we were to go back to "again" then we may never
		 * catch up, and will trigger the warn on, or lock
		 * the box. Return the padding, and we will release
		 * the current locks, and try again.
		 */
		return event;
S
Steven Rostedt 已提交
3440 3441 3442

	case RINGBUF_TYPE_TIME_EXTEND:
		/* Internal data, OK to advance */
3443
		rb_advance_reader(cpu_buffer);
S
Steven Rostedt 已提交
3444 3445 3446 3447
		goto again;

	case RINGBUF_TYPE_TIME_STAMP:
		/* FIXME: not implemented */
3448
		rb_advance_reader(cpu_buffer);
S
Steven Rostedt 已提交
3449 3450 3451 3452 3453
		goto again;

	case RINGBUF_TYPE_DATA:
		if (ts) {
			*ts = cpu_buffer->read_stamp + event->time_delta;
3454
			ring_buffer_normalize_time_stamp(cpu_buffer->buffer,
3455
							 cpu_buffer->cpu, ts);
S
Steven Rostedt 已提交
3456
		}
3457 3458
		if (lost_events)
			*lost_events = rb_lost_events(cpu_buffer);
S
Steven Rostedt 已提交
3459 3460 3461 3462 3463 3464 3465 3466
		return event;

	default:
		BUG();
	}

	return NULL;
}
3467
EXPORT_SYMBOL_GPL(ring_buffer_peek);
S
Steven Rostedt 已提交
3468

S
Steven Rostedt 已提交
3469 3470
static struct ring_buffer_event *
rb_iter_peek(struct ring_buffer_iter *iter, u64 *ts)
S
Steven Rostedt 已提交
3471 3472 3473 3474
{
	struct ring_buffer *buffer;
	struct ring_buffer_per_cpu *cpu_buffer;
	struct ring_buffer_event *event;
3475
	int nr_loops = 0;
S
Steven Rostedt 已提交
3476 3477 3478 3479

	cpu_buffer = iter->cpu_buffer;
	buffer = cpu_buffer->buffer;

3480 3481 3482 3483 3484 3485 3486 3487 3488
	/*
	 * Check if someone performed a consuming read to
	 * the buffer. A consuming read invalidates the iterator
	 * and we need to reset the iterator in this case.
	 */
	if (unlikely(iter->cache_read != cpu_buffer->read ||
		     iter->cache_reader_page != cpu_buffer->reader_page))
		rb_iter_reset(iter);

S
Steven Rostedt 已提交
3489
 again:
3490 3491 3492
	if (ring_buffer_iter_empty(iter))
		return NULL;

3493
	/*
3494 3495 3496 3497
	 * We repeat when a time extend is encountered.
	 * Since the time extend is always attached to a data event,
	 * we should never loop more than once.
	 * (We never hit the following condition more than twice).
3498
	 */
3499
	if (RB_WARN_ON(cpu_buffer, ++nr_loops > 2))
3500 3501
		return NULL;

S
Steven Rostedt 已提交
3502 3503 3504
	if (rb_per_cpu_empty(cpu_buffer))
		return NULL;

3505 3506 3507 3508 3509
	if (iter->head >= local_read(&iter->head_page->page->commit)) {
		rb_inc_iter(iter);
		goto again;
	}

S
Steven Rostedt 已提交
3510 3511
	event = rb_iter_head_event(iter);

3512
	switch (event->type_len) {
S
Steven Rostedt 已提交
3513
	case RINGBUF_TYPE_PADDING:
3514 3515 3516 3517 3518 3519
		if (rb_null_event(event)) {
			rb_inc_iter(iter);
			goto again;
		}
		rb_advance_iter(iter);
		return event;
S
Steven Rostedt 已提交
3520 3521 3522 3523 3524 3525 3526 3527 3528 3529 3530 3531 3532 3533

	case RINGBUF_TYPE_TIME_EXTEND:
		/* Internal data, OK to advance */
		rb_advance_iter(iter);
		goto again;

	case RINGBUF_TYPE_TIME_STAMP:
		/* FIXME: not implemented */
		rb_advance_iter(iter);
		goto again;

	case RINGBUF_TYPE_DATA:
		if (ts) {
			*ts = iter->read_stamp + event->time_delta;
3534 3535
			ring_buffer_normalize_time_stamp(buffer,
							 cpu_buffer->cpu, ts);
S
Steven Rostedt 已提交
3536 3537 3538 3539 3540 3541 3542 3543 3544
		}
		return event;

	default:
		BUG();
	}

	return NULL;
}
3545
EXPORT_SYMBOL_GPL(ring_buffer_iter_peek);
S
Steven Rostedt 已提交
3546

3547 3548 3549 3550 3551 3552 3553 3554
static inline int rb_ok_to_lock(void)
{
	/*
	 * If an NMI die dumps out the content of the ring buffer
	 * do not grab locks. We also permanently disable the ring
	 * buffer too. A one time deal is all you get from reading
	 * the ring buffer from an NMI.
	 */
3555
	if (likely(!in_nmi()))
3556 3557 3558 3559 3560 3561
		return 1;

	tracing_off_permanent();
	return 0;
}

S
Steven Rostedt 已提交
3562 3563 3564 3565 3566
/**
 * ring_buffer_peek - peek at the next event to be read
 * @buffer: The ring buffer to read
 * @cpu: The cpu to peak at
 * @ts: The timestamp counter of this event.
3567
 * @lost_events: a variable to store if events were lost (may be NULL)
S
Steven Rostedt 已提交
3568 3569 3570 3571 3572
 *
 * This will return the event that will be read next, but does
 * not consume the data.
 */
struct ring_buffer_event *
3573 3574
ring_buffer_peek(struct ring_buffer *buffer, int cpu, u64 *ts,
		 unsigned long *lost_events)
S
Steven Rostedt 已提交
3575 3576
{
	struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
3577
	struct ring_buffer_event *event;
S
Steven Rostedt 已提交
3578
	unsigned long flags;
3579
	int dolock;
S
Steven Rostedt 已提交
3580

3581
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
3582
		return NULL;
3583

3584
	dolock = rb_ok_to_lock();
3585
 again:
3586 3587
	local_irq_save(flags);
	if (dolock)
3588
		raw_spin_lock(&cpu_buffer->reader_lock);
3589
	event = rb_buffer_peek(cpu_buffer, ts, lost_events);
3590 3591
	if (event && event->type_len == RINGBUF_TYPE_PADDING)
		rb_advance_reader(cpu_buffer);
3592
	if (dolock)
3593
		raw_spin_unlock(&cpu_buffer->reader_lock);
3594
	local_irq_restore(flags);
S
Steven Rostedt 已提交
3595

3596
	if (event && event->type_len == RINGBUF_TYPE_PADDING)
3597 3598
		goto again;

S
Steven Rostedt 已提交
3599 3600 3601 3602 3603 3604 3605 3606 3607 3608 3609 3610 3611 3612 3613 3614 3615 3616
	return event;
}

/**
 * ring_buffer_iter_peek - peek at the next event to be read
 * @iter: The ring buffer iterator
 * @ts: The timestamp counter of this event.
 *
 * This will return the event that will be read next, but does
 * not increment the iterator.
 */
struct ring_buffer_event *
ring_buffer_iter_peek(struct ring_buffer_iter *iter, u64 *ts)
{
	struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
	struct ring_buffer_event *event;
	unsigned long flags;

3617
 again:
3618
	raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
S
Steven Rostedt 已提交
3619
	event = rb_iter_peek(iter, ts);
3620
	raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
S
Steven Rostedt 已提交
3621

3622
	if (event && event->type_len == RINGBUF_TYPE_PADDING)
3623 3624
		goto again;

S
Steven Rostedt 已提交
3625 3626 3627
	return event;
}

S
Steven Rostedt 已提交
3628 3629 3630
/**
 * ring_buffer_consume - return an event and consume it
 * @buffer: The ring buffer to get the next event from
3631 3632 3633
 * @cpu: the cpu to read the buffer from
 * @ts: a variable to store the timestamp (may be NULL)
 * @lost_events: a variable to store if events were lost (may be NULL)
S
Steven Rostedt 已提交
3634 3635 3636 3637 3638 3639
 *
 * Returns the next event in the ring buffer, and that event is consumed.
 * Meaning, that sequential reads will keep returning a different event,
 * and eventually empty the ring buffer if the producer is slower.
 */
struct ring_buffer_event *
3640 3641
ring_buffer_consume(struct ring_buffer *buffer, int cpu, u64 *ts,
		    unsigned long *lost_events)
S
Steven Rostedt 已提交
3642
{
3643 3644
	struct ring_buffer_per_cpu *cpu_buffer;
	struct ring_buffer_event *event = NULL;
S
Steven Rostedt 已提交
3645
	unsigned long flags;
3646 3647 3648
	int dolock;

	dolock = rb_ok_to_lock();
S
Steven Rostedt 已提交
3649

3650
 again:
3651 3652 3653
	/* might be called in atomic */
	preempt_disable();

3654
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
3655
		goto out;
S
Steven Rostedt 已提交
3656

3657
	cpu_buffer = buffer->buffers[cpu];
3658 3659
	local_irq_save(flags);
	if (dolock)
3660
		raw_spin_lock(&cpu_buffer->reader_lock);
S
Steven Rostedt 已提交
3661

3662 3663 3664
	event = rb_buffer_peek(cpu_buffer, ts, lost_events);
	if (event) {
		cpu_buffer->lost_events = 0;
3665
		rb_advance_reader(cpu_buffer);
3666
	}
S
Steven Rostedt 已提交
3667

3668
	if (dolock)
3669
		raw_spin_unlock(&cpu_buffer->reader_lock);
3670
	local_irq_restore(flags);
S
Steven Rostedt 已提交
3671

3672 3673 3674
 out:
	preempt_enable();

3675
	if (event && event->type_len == RINGBUF_TYPE_PADDING)
3676 3677
		goto again;

S
Steven Rostedt 已提交
3678 3679
	return event;
}
3680
EXPORT_SYMBOL_GPL(ring_buffer_consume);
S
Steven Rostedt 已提交
3681 3682

/**
3683
 * ring_buffer_read_prepare - Prepare for a non consuming read of the buffer
S
Steven Rostedt 已提交
3684 3685 3686
 * @buffer: The ring buffer to read from
 * @cpu: The cpu buffer to iterate over
 *
3687 3688 3689
 * This performs the initial preparations necessary to iterate
 * through the buffer.  Memory is allocated, buffer recording
 * is disabled, and the iterator pointer is returned to the caller.
S
Steven Rostedt 已提交
3690
 *
3691 3692 3693 3694 3695 3696 3697 3698 3699 3700
 * Disabling buffer recordng prevents the reading from being
 * corrupted. This is not a consuming read, so a producer is not
 * expected.
 *
 * After a sequence of ring_buffer_read_prepare calls, the user is
 * expected to make at least one call to ring_buffer_prepare_sync.
 * Afterwards, ring_buffer_read_start is invoked to get things going
 * for real.
 *
 * This overall must be paired with ring_buffer_finish.
S
Steven Rostedt 已提交
3701 3702
 */
struct ring_buffer_iter *
3703
ring_buffer_read_prepare(struct ring_buffer *buffer, int cpu)
S
Steven Rostedt 已提交
3704 3705
{
	struct ring_buffer_per_cpu *cpu_buffer;
3706
	struct ring_buffer_iter *iter;
S
Steven Rostedt 已提交
3707

3708
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
3709
		return NULL;
S
Steven Rostedt 已提交
3710 3711 3712

	iter = kmalloc(sizeof(*iter), GFP_KERNEL);
	if (!iter)
3713
		return NULL;
S
Steven Rostedt 已提交
3714 3715 3716 3717 3718

	cpu_buffer = buffer->buffers[cpu];

	iter->cpu_buffer = cpu_buffer;

3719
	atomic_inc(&buffer->resize_disabled);
S
Steven Rostedt 已提交
3720
	atomic_inc(&cpu_buffer->record_disabled);
3721 3722 3723 3724 3725 3726 3727 3728 3729 3730 3731 3732 3733 3734 3735

	return iter;
}
EXPORT_SYMBOL_GPL(ring_buffer_read_prepare);

/**
 * ring_buffer_read_prepare_sync - Synchronize a set of prepare calls
 *
 * All previously invoked ring_buffer_read_prepare calls to prepare
 * iterators will be synchronized.  Afterwards, read_buffer_read_start
 * calls on those iterators are allowed.
 */
void
ring_buffer_read_prepare_sync(void)
{
S
Steven Rostedt 已提交
3736
	synchronize_sched();
3737 3738 3739 3740 3741 3742 3743 3744 3745 3746 3747 3748 3749 3750 3751 3752 3753 3754 3755 3756 3757 3758 3759 3760
}
EXPORT_SYMBOL_GPL(ring_buffer_read_prepare_sync);

/**
 * ring_buffer_read_start - start a non consuming read of the buffer
 * @iter: The iterator returned by ring_buffer_read_prepare
 *
 * This finalizes the startup of an iteration through the buffer.
 * The iterator comes from a call to ring_buffer_read_prepare and
 * an intervening ring_buffer_read_prepare_sync must have been
 * performed.
 *
 * Must be paired with ring_buffer_finish.
 */
void
ring_buffer_read_start(struct ring_buffer_iter *iter)
{
	struct ring_buffer_per_cpu *cpu_buffer;
	unsigned long flags;

	if (!iter)
		return;

	cpu_buffer = iter->cpu_buffer;
S
Steven Rostedt 已提交
3761

3762
	raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
3763
	arch_spin_lock(&cpu_buffer->lock);
3764
	rb_iter_reset(iter);
3765
	arch_spin_unlock(&cpu_buffer->lock);
3766
	raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
S
Steven Rostedt 已提交
3767
}
3768
EXPORT_SYMBOL_GPL(ring_buffer_read_start);
S
Steven Rostedt 已提交
3769 3770 3771 3772 3773 3774 3775 3776 3777 3778 3779 3780 3781

/**
 * ring_buffer_finish - finish reading the iterator of the buffer
 * @iter: The iterator retrieved by ring_buffer_start
 *
 * This re-enables the recording to the buffer, and frees the
 * iterator.
 */
void
ring_buffer_read_finish(struct ring_buffer_iter *iter)
{
	struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;

3782 3783 3784 3785 3786 3787
	/*
	 * Ring buffer is disabled from recording, here's a good place
	 * to check the integrity of the ring buffer. 
	 */
	rb_check_pages(cpu_buffer);

S
Steven Rostedt 已提交
3788
	atomic_dec(&cpu_buffer->record_disabled);
3789
	atomic_dec(&cpu_buffer->buffer->resize_disabled);
S
Steven Rostedt 已提交
3790 3791
	kfree(iter);
}
3792
EXPORT_SYMBOL_GPL(ring_buffer_read_finish);
S
Steven Rostedt 已提交
3793 3794 3795 3796 3797 3798 3799 3800 3801 3802 3803 3804

/**
 * ring_buffer_read - read the next item in the ring buffer by the iterator
 * @iter: The ring buffer iterator
 * @ts: The time stamp of the event read.
 *
 * This reads the next event in the ring buffer and increments the iterator.
 */
struct ring_buffer_event *
ring_buffer_read(struct ring_buffer_iter *iter, u64 *ts)
{
	struct ring_buffer_event *event;
S
Steven Rostedt 已提交
3805 3806
	struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
	unsigned long flags;
S
Steven Rostedt 已提交
3807

3808
	raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
3809
 again:
S
Steven Rostedt 已提交
3810
	event = rb_iter_peek(iter, ts);
S
Steven Rostedt 已提交
3811
	if (!event)
S
Steven Rostedt 已提交
3812
		goto out;
S
Steven Rostedt 已提交
3813

3814 3815 3816
	if (event->type_len == RINGBUF_TYPE_PADDING)
		goto again;

S
Steven Rostedt 已提交
3817
	rb_advance_iter(iter);
S
Steven Rostedt 已提交
3818
 out:
3819
	raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
S
Steven Rostedt 已提交
3820 3821 3822

	return event;
}
3823
EXPORT_SYMBOL_GPL(ring_buffer_read);
S
Steven Rostedt 已提交
3824 3825 3826 3827 3828

/**
 * ring_buffer_size - return the size of the ring buffer (in bytes)
 * @buffer: The ring buffer.
 */
3829
unsigned long ring_buffer_size(struct ring_buffer *buffer, int cpu)
S
Steven Rostedt 已提交
3830
{
3831 3832 3833 3834 3835 3836 3837 3838 3839 3840
	/*
	 * Earlier, this method returned
	 *	BUF_PAGE_SIZE * buffer->nr_pages
	 * Since the nr_pages field is now removed, we have converted this to
	 * return the per cpu buffer value.
	 */
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
		return 0;

	return BUF_PAGE_SIZE * buffer->buffers[cpu]->nr_pages;
S
Steven Rostedt 已提交
3841
}
3842
EXPORT_SYMBOL_GPL(ring_buffer_size);
S
Steven Rostedt 已提交
3843 3844 3845 3846

static void
rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer)
{
S
Steven Rostedt 已提交
3847 3848
	rb_head_page_deactivate(cpu_buffer);

S
Steven Rostedt 已提交
3849
	cpu_buffer->head_page
3850
		= list_entry(cpu_buffer->pages, struct buffer_page, list);
S
Steven Rostedt 已提交
3851
	local_set(&cpu_buffer->head_page->write, 0);
3852
	local_set(&cpu_buffer->head_page->entries, 0);
3853
	local_set(&cpu_buffer->head_page->page->commit, 0);
3854

3855
	cpu_buffer->head_page->read = 0;
S
Steven Rostedt 已提交
3856 3857 3858 3859 3860

	cpu_buffer->tail_page = cpu_buffer->head_page;
	cpu_buffer->commit_page = cpu_buffer->head_page;

	INIT_LIST_HEAD(&cpu_buffer->reader_page->list);
3861
	INIT_LIST_HEAD(&cpu_buffer->new_pages);
S
Steven Rostedt 已提交
3862
	local_set(&cpu_buffer->reader_page->write, 0);
3863
	local_set(&cpu_buffer->reader_page->entries, 0);
3864
	local_set(&cpu_buffer->reader_page->page->commit, 0);
3865
	cpu_buffer->reader_page->read = 0;
S
Steven Rostedt 已提交
3866

S
Steven Rostedt 已提交
3867
	local_set(&cpu_buffer->commit_overrun, 0);
3868
	local_set(&cpu_buffer->entries_bytes, 0);
S
Steven Rostedt 已提交
3869
	local_set(&cpu_buffer->overrun, 0);
3870
	local_set(&cpu_buffer->entries, 0);
3871 3872
	local_set(&cpu_buffer->committing, 0);
	local_set(&cpu_buffer->commits, 0);
S
Steven Rostedt 已提交
3873
	cpu_buffer->read = 0;
3874
	cpu_buffer->read_bytes = 0;
3875 3876 3877

	cpu_buffer->write_stamp = 0;
	cpu_buffer->read_stamp = 0;
S
Steven Rostedt 已提交
3878

3879 3880 3881
	cpu_buffer->lost_events = 0;
	cpu_buffer->last_overrun = 0;

S
Steven Rostedt 已提交
3882
	rb_head_page_activate(cpu_buffer);
S
Steven Rostedt 已提交
3883 3884 3885 3886 3887 3888 3889 3890 3891 3892 3893 3894
}

/**
 * ring_buffer_reset_cpu - reset a ring buffer per CPU buffer
 * @buffer: The ring buffer to reset a per cpu buffer of
 * @cpu: The CPU buffer to be reset
 */
void ring_buffer_reset_cpu(struct ring_buffer *buffer, int cpu)
{
	struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
	unsigned long flags;

3895
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
3896
		return;
S
Steven Rostedt 已提交
3897

3898
	atomic_inc(&buffer->resize_disabled);
3899 3900
	atomic_inc(&cpu_buffer->record_disabled);

3901 3902 3903
	/* Make sure all commits have finished */
	synchronize_sched();

3904
	raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
S
Steven Rostedt 已提交
3905

3906 3907 3908
	if (RB_WARN_ON(cpu_buffer, local_read(&cpu_buffer->committing)))
		goto out;

3909
	arch_spin_lock(&cpu_buffer->lock);
S
Steven Rostedt 已提交
3910 3911 3912

	rb_reset_cpu(cpu_buffer);

3913
	arch_spin_unlock(&cpu_buffer->lock);
S
Steven Rostedt 已提交
3914

3915
 out:
3916
	raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
3917 3918

	atomic_dec(&cpu_buffer->record_disabled);
3919
	atomic_dec(&buffer->resize_disabled);
S
Steven Rostedt 已提交
3920
}
3921
EXPORT_SYMBOL_GPL(ring_buffer_reset_cpu);
S
Steven Rostedt 已提交
3922 3923 3924 3925 3926 3927 3928 3929 3930 3931

/**
 * ring_buffer_reset - reset a ring buffer
 * @buffer: The ring buffer to reset all cpu buffers
 */
void ring_buffer_reset(struct ring_buffer *buffer)
{
	int cpu;

	for_each_buffer_cpu(buffer, cpu)
3932
		ring_buffer_reset_cpu(buffer, cpu);
S
Steven Rostedt 已提交
3933
}
3934
EXPORT_SYMBOL_GPL(ring_buffer_reset);
S
Steven Rostedt 已提交
3935 3936 3937 3938 3939 3940 3941 3942

/**
 * rind_buffer_empty - is the ring buffer empty?
 * @buffer: The ring buffer to test
 */
int ring_buffer_empty(struct ring_buffer *buffer)
{
	struct ring_buffer_per_cpu *cpu_buffer;
3943
	unsigned long flags;
3944
	int dolock;
S
Steven Rostedt 已提交
3945
	int cpu;
3946
	int ret;
S
Steven Rostedt 已提交
3947

3948
	dolock = rb_ok_to_lock();
S
Steven Rostedt 已提交
3949 3950 3951 3952

	/* yes this is racy, but if you don't like the race, lock the buffer */
	for_each_buffer_cpu(buffer, cpu) {
		cpu_buffer = buffer->buffers[cpu];
3953 3954
		local_irq_save(flags);
		if (dolock)
3955
			raw_spin_lock(&cpu_buffer->reader_lock);
3956
		ret = rb_per_cpu_empty(cpu_buffer);
3957
		if (dolock)
3958
			raw_spin_unlock(&cpu_buffer->reader_lock);
3959 3960
		local_irq_restore(flags);

3961
		if (!ret)
S
Steven Rostedt 已提交
3962 3963
			return 0;
	}
3964

S
Steven Rostedt 已提交
3965 3966
	return 1;
}
3967
EXPORT_SYMBOL_GPL(ring_buffer_empty);
S
Steven Rostedt 已提交
3968 3969 3970 3971 3972 3973 3974 3975 3976

/**
 * ring_buffer_empty_cpu - is a cpu buffer of a ring buffer empty?
 * @buffer: The ring buffer
 * @cpu: The CPU buffer to test
 */
int ring_buffer_empty_cpu(struct ring_buffer *buffer, int cpu)
{
	struct ring_buffer_per_cpu *cpu_buffer;
3977
	unsigned long flags;
3978
	int dolock;
3979
	int ret;
S
Steven Rostedt 已提交
3980

3981
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
3982
		return 1;
S
Steven Rostedt 已提交
3983

3984 3985
	dolock = rb_ok_to_lock();

S
Steven Rostedt 已提交
3986
	cpu_buffer = buffer->buffers[cpu];
3987 3988
	local_irq_save(flags);
	if (dolock)
3989
		raw_spin_lock(&cpu_buffer->reader_lock);
3990
	ret = rb_per_cpu_empty(cpu_buffer);
3991
	if (dolock)
3992
		raw_spin_unlock(&cpu_buffer->reader_lock);
3993
	local_irq_restore(flags);
3994 3995

	return ret;
S
Steven Rostedt 已提交
3996
}
3997
EXPORT_SYMBOL_GPL(ring_buffer_empty_cpu);
S
Steven Rostedt 已提交
3998

3999
#ifdef CONFIG_RING_BUFFER_ALLOW_SWAP
S
Steven Rostedt 已提交
4000 4001 4002 4003 4004 4005 4006 4007 4008 4009 4010 4011 4012 4013 4014
/**
 * ring_buffer_swap_cpu - swap a CPU buffer between two ring buffers
 * @buffer_a: One buffer to swap with
 * @buffer_b: The other buffer to swap with
 *
 * This function is useful for tracers that want to take a "snapshot"
 * of a CPU buffer and has another back up buffer lying around.
 * it is expected that the tracer handles the cpu buffer not being
 * used at the moment.
 */
int ring_buffer_swap_cpu(struct ring_buffer *buffer_a,
			 struct ring_buffer *buffer_b, int cpu)
{
	struct ring_buffer_per_cpu *cpu_buffer_a;
	struct ring_buffer_per_cpu *cpu_buffer_b;
4015 4016
	int ret = -EINVAL;

4017 4018
	if (!cpumask_test_cpu(cpu, buffer_a->cpumask) ||
	    !cpumask_test_cpu(cpu, buffer_b->cpumask))
4019
		goto out;
S
Steven Rostedt 已提交
4020

4021 4022 4023
	cpu_buffer_a = buffer_a->buffers[cpu];
	cpu_buffer_b = buffer_b->buffers[cpu];

S
Steven Rostedt 已提交
4024
	/* At least make sure the two buffers are somewhat the same */
4025
	if (cpu_buffer_a->nr_pages != cpu_buffer_b->nr_pages)
4026 4027 4028
		goto out;

	ret = -EAGAIN;
S
Steven Rostedt 已提交
4029

4030
	if (ring_buffer_flags != RB_BUFFERS_ON)
4031
		goto out;
4032 4033

	if (atomic_read(&buffer_a->record_disabled))
4034
		goto out;
4035 4036

	if (atomic_read(&buffer_b->record_disabled))
4037
		goto out;
4038 4039

	if (atomic_read(&cpu_buffer_a->record_disabled))
4040
		goto out;
4041 4042

	if (atomic_read(&cpu_buffer_b->record_disabled))
4043
		goto out;
4044

S
Steven Rostedt 已提交
4045 4046 4047 4048 4049 4050 4051 4052 4053
	/*
	 * We can't do a synchronize_sched here because this
	 * function can be called in atomic context.
	 * Normally this will be called from the same CPU as cpu.
	 * If not it's up to the caller to protect this.
	 */
	atomic_inc(&cpu_buffer_a->record_disabled);
	atomic_inc(&cpu_buffer_b->record_disabled);

4054 4055 4056 4057 4058 4059
	ret = -EBUSY;
	if (local_read(&cpu_buffer_a->committing))
		goto out_dec;
	if (local_read(&cpu_buffer_b->committing))
		goto out_dec;

S
Steven Rostedt 已提交
4060 4061 4062 4063 4064 4065
	buffer_a->buffers[cpu] = cpu_buffer_b;
	buffer_b->buffers[cpu] = cpu_buffer_a;

	cpu_buffer_b->buffer = buffer_a;
	cpu_buffer_a->buffer = buffer_b;

4066 4067 4068
	ret = 0;

out_dec:
S
Steven Rostedt 已提交
4069 4070
	atomic_dec(&cpu_buffer_a->record_disabled);
	atomic_dec(&cpu_buffer_b->record_disabled);
4071 4072
out:
	return ret;
S
Steven Rostedt 已提交
4073
}
4074
EXPORT_SYMBOL_GPL(ring_buffer_swap_cpu);
4075
#endif /* CONFIG_RING_BUFFER_ALLOW_SWAP */
S
Steven Rostedt 已提交
4076

S
Steven Rostedt 已提交
4077 4078 4079 4080 4081 4082 4083 4084 4085 4086 4087 4088 4089 4090 4091
/**
 * ring_buffer_alloc_read_page - allocate a page to read from buffer
 * @buffer: the buffer to allocate for.
 *
 * This function is used in conjunction with ring_buffer_read_page.
 * When reading a full page from the ring buffer, these functions
 * can be used to speed up the process. The calling function should
 * allocate a few pages first with this function. Then when it
 * needs to get pages from the ring buffer, it passes the result
 * of this function into ring_buffer_read_page, which will swap
 * the page that was allocated, with the read page of the buffer.
 *
 * Returns:
 *  The page allocated, or NULL on error.
 */
4092
void *ring_buffer_alloc_read_page(struct ring_buffer *buffer, int cpu)
S
Steven Rostedt 已提交
4093
{
4094
	struct buffer_data_page *bpage;
4095
	struct page *page;
S
Steven Rostedt 已提交
4096

4097 4098
	page = alloc_pages_node(cpu_to_node(cpu),
				GFP_KERNEL | __GFP_NORETRY, 0);
4099
	if (!page)
S
Steven Rostedt 已提交
4100 4101
		return NULL;

4102
	bpage = page_address(page);
S
Steven Rostedt 已提交
4103

4104 4105
	rb_init_page(bpage);

4106
	return bpage;
S
Steven Rostedt 已提交
4107
}
S
Steven Rostedt 已提交
4108
EXPORT_SYMBOL_GPL(ring_buffer_alloc_read_page);
S
Steven Rostedt 已提交
4109 4110 4111 4112 4113 4114 4115 4116 4117 4118 4119 4120

/**
 * ring_buffer_free_read_page - free an allocated read page
 * @buffer: the buffer the page was allocate for
 * @data: the page to free
 *
 * Free a page allocated from ring_buffer_alloc_read_page.
 */
void ring_buffer_free_read_page(struct ring_buffer *buffer, void *data)
{
	free_page((unsigned long)data);
}
S
Steven Rostedt 已提交
4121
EXPORT_SYMBOL_GPL(ring_buffer_free_read_page);
S
Steven Rostedt 已提交
4122 4123 4124 4125 4126

/**
 * ring_buffer_read_page - extract a page from the ring buffer
 * @buffer: buffer to extract from
 * @data_page: the page to use allocated from ring_buffer_alloc_read_page
4127
 * @len: amount to extract
S
Steven Rostedt 已提交
4128 4129 4130 4131 4132 4133 4134 4135 4136
 * @cpu: the cpu of the buffer to extract
 * @full: should the extraction only happen when the page is full.
 *
 * This function will pull out a page from the ring buffer and consume it.
 * @data_page must be the address of the variable that was returned
 * from ring_buffer_alloc_read_page. This is because the page might be used
 * to swap with a page in the ring buffer.
 *
 * for example:
L
Lai Jiangshan 已提交
4137
 *	rpage = ring_buffer_alloc_read_page(buffer);
S
Steven Rostedt 已提交
4138 4139
 *	if (!rpage)
 *		return error;
4140
 *	ret = ring_buffer_read_page(buffer, &rpage, len, cpu, 0);
4141 4142
 *	if (ret >= 0)
 *		process_page(rpage, ret);
S
Steven Rostedt 已提交
4143 4144 4145 4146 4147 4148 4149 4150 4151 4152
 *
 * When @full is set, the function will not return true unless
 * the writer is off the reader page.
 *
 * Note: it is up to the calling functions to handle sleeps and wakeups.
 *  The ring buffer can be used anywhere in the kernel and can not
 *  blindly call wake_up. The layer that uses the ring buffer must be
 *  responsible for that.
 *
 * Returns:
4153 4154
 *  >=0 if data has been transferred, returns the offset of consumed data.
 *  <0 if no data has been transferred.
S
Steven Rostedt 已提交
4155 4156
 */
int ring_buffer_read_page(struct ring_buffer *buffer,
4157
			  void **data_page, size_t len, int cpu, int full)
S
Steven Rostedt 已提交
4158 4159 4160
{
	struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
	struct ring_buffer_event *event;
4161
	struct buffer_data_page *bpage;
4162
	struct buffer_page *reader;
4163
	unsigned long missed_events;
S
Steven Rostedt 已提交
4164
	unsigned long flags;
4165
	unsigned int commit;
4166
	unsigned int read;
4167
	u64 save_timestamp;
4168
	int ret = -1;
S
Steven Rostedt 已提交
4169

4170 4171 4172
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
		goto out;

4173 4174 4175 4176 4177
	/*
	 * If len is not big enough to hold the page header, then
	 * we can not copy anything.
	 */
	if (len <= BUF_PAGE_HDR_SIZE)
4178
		goto out;
4179 4180 4181

	len -= BUF_PAGE_HDR_SIZE;

S
Steven Rostedt 已提交
4182
	if (!data_page)
4183
		goto out;
S
Steven Rostedt 已提交
4184

4185 4186
	bpage = *data_page;
	if (!bpage)
4187
		goto out;
S
Steven Rostedt 已提交
4188

4189
	raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
S
Steven Rostedt 已提交
4190

4191 4192
	reader = rb_get_reader_page(cpu_buffer);
	if (!reader)
4193
		goto out_unlock;
S
Steven Rostedt 已提交
4194

4195 4196 4197 4198
	event = rb_reader_event(cpu_buffer);

	read = reader->read;
	commit = rb_page_commit(reader);
4199

4200
	/* Check if any events were dropped */
4201
	missed_events = cpu_buffer->lost_events;
4202

S
Steven Rostedt 已提交
4203
	/*
4204 4205 4206 4207 4208
	 * If this page has been partially read or
	 * if len is not big enough to read the rest of the page or
	 * a writer is still on the page, then
	 * we must copy the data from the page to the buffer.
	 * Otherwise, we can simply swap the page with the one passed in.
S
Steven Rostedt 已提交
4209
	 */
4210
	if (read || (len < (commit - read)) ||
4211
	    cpu_buffer->reader_page == cpu_buffer->commit_page) {
4212
		struct buffer_data_page *rpage = cpu_buffer->reader_page->page;
4213 4214
		unsigned int rpos = read;
		unsigned int pos = 0;
4215
		unsigned int size;
S
Steven Rostedt 已提交
4216 4217

		if (full)
4218
			goto out_unlock;
S
Steven Rostedt 已提交
4219

4220 4221 4222
		if (len > (commit - read))
			len = (commit - read);

4223 4224
		/* Always keep the time extend and data together */
		size = rb_event_ts_length(event);
4225 4226

		if (len < size)
4227
			goto out_unlock;
4228

4229 4230 4231
		/* save the current timestamp, since the user will need it */
		save_timestamp = cpu_buffer->read_stamp;

4232 4233
		/* Need to copy one event at a time */
		do {
4234 4235 4236 4237 4238 4239 4240
			/* We need the size of one event, because
			 * rb_advance_reader only advances by one event,
			 * whereas rb_event_ts_length may include the size of
			 * one or two events.
			 * We have already ensured there's enough space if this
			 * is a time extend. */
			size = rb_event_length(event);
4241
			memcpy(bpage->data + pos, rpage->data + rpos, size);
4242 4243 4244 4245

			len -= size;

			rb_advance_reader(cpu_buffer);
4246 4247
			rpos = reader->read;
			pos += size;
4248

4249 4250 4251
			if (rpos >= commit)
				break;

4252
			event = rb_reader_event(cpu_buffer);
4253 4254
			/* Always keep the time extend and data together */
			size = rb_event_ts_length(event);
4255
		} while (len >= size);
4256 4257

		/* update bpage */
4258
		local_set(&bpage->commit, pos);
4259
		bpage->time_stamp = save_timestamp;
4260

4261 4262
		/* we copied everything to the beginning */
		read = 0;
S
Steven Rostedt 已提交
4263
	} else {
4264
		/* update the entry counter */
S
Steven Rostedt 已提交
4265
		cpu_buffer->read += rb_page_entries(reader);
4266
		cpu_buffer->read_bytes += BUF_PAGE_SIZE;
4267

S
Steven Rostedt 已提交
4268
		/* swap the pages */
4269
		rb_init_page(bpage);
4270 4271 4272
		bpage = reader->page;
		reader->page = *data_page;
		local_set(&reader->write, 0);
4273
		local_set(&reader->entries, 0);
4274
		reader->read = 0;
4275
		*data_page = bpage;
4276 4277 4278 4279 4280 4281 4282 4283

		/*
		 * Use the real_end for the data size,
		 * This gives us a chance to store the lost events
		 * on the page.
		 */
		if (reader->real_end)
			local_set(&bpage->commit, reader->real_end);
S
Steven Rostedt 已提交
4284
	}
4285
	ret = read;
S
Steven Rostedt 已提交
4286

4287
	cpu_buffer->lost_events = 0;
4288 4289

	commit = local_read(&bpage->commit);
4290 4291 4292
	/*
	 * Set a flag in the commit field if we lost events
	 */
4293 4294 4295 4296 4297 4298 4299 4300
	if (missed_events) {
		/* If there is room at the end of the page to save the
		 * missed events, then record it there.
		 */
		if (BUF_PAGE_SIZE - commit >= sizeof(missed_events)) {
			memcpy(&bpage->data[commit], &missed_events,
			       sizeof(missed_events));
			local_add(RB_MISSED_STORED, &bpage->commit);
4301
			commit += sizeof(missed_events);
4302
		}
4303
		local_add(RB_MISSED_EVENTS, &bpage->commit);
4304
	}
4305

4306 4307 4308 4309 4310 4311
	/*
	 * This page may be off to user land. Zero it out here.
	 */
	if (commit < BUF_PAGE_SIZE)
		memset(&bpage->data[commit], 0, BUF_PAGE_SIZE - commit);

4312
 out_unlock:
4313
	raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
S
Steven Rostedt 已提交
4314

4315
 out:
S
Steven Rostedt 已提交
4316 4317
	return ret;
}
S
Steven Rostedt 已提交
4318
EXPORT_SYMBOL_GPL(ring_buffer_read_page);
S
Steven Rostedt 已提交
4319

4320
#ifdef CONFIG_HOTPLUG_CPU
4321 4322
static int rb_cpu_notify(struct notifier_block *self,
			 unsigned long action, void *hcpu)
4323 4324 4325 4326
{
	struct ring_buffer *buffer =
		container_of(self, struct ring_buffer, cpu_notify);
	long cpu = (long)hcpu;
4327 4328
	int cpu_i, nr_pages_same;
	unsigned int nr_pages;
4329 4330 4331 4332

	switch (action) {
	case CPU_UP_PREPARE:
	case CPU_UP_PREPARE_FROZEN:
4333
		if (cpumask_test_cpu(cpu, buffer->cpumask))
4334 4335
			return NOTIFY_OK;

4336 4337 4338 4339 4340 4341 4342 4343 4344 4345 4346 4347 4348 4349 4350
		nr_pages = 0;
		nr_pages_same = 1;
		/* check if all cpu sizes are same */
		for_each_buffer_cpu(buffer, cpu_i) {
			/* fill in the size from first enabled cpu */
			if (nr_pages == 0)
				nr_pages = buffer->buffers[cpu_i]->nr_pages;
			if (nr_pages != buffer->buffers[cpu_i]->nr_pages) {
				nr_pages_same = 0;
				break;
			}
		}
		/* allocate minimum pages, user can later expand it */
		if (!nr_pages_same)
			nr_pages = 2;
4351
		buffer->buffers[cpu] =
4352
			rb_allocate_cpu_buffer(buffer, nr_pages, cpu);
4353 4354 4355 4356 4357 4358
		if (!buffer->buffers[cpu]) {
			WARN(1, "failed to allocate ring buffer on CPU %ld\n",
			     cpu);
			return NOTIFY_OK;
		}
		smp_wmb();
4359
		cpumask_set_cpu(cpu, buffer->cpumask);
4360 4361 4362 4363 4364 4365 4366 4367 4368 4369 4370 4371 4372 4373 4374
		break;
	case CPU_DOWN_PREPARE:
	case CPU_DOWN_PREPARE_FROZEN:
		/*
		 * Do nothing.
		 *  If we were to free the buffer, then the user would
		 *  lose any trace that was in the buffer.
		 */
		break;
	default:
		break;
	}
	return NOTIFY_OK;
}
#endif