ring_buffer.c 96.0 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>
8
#include <linux/ftrace_irq.h>
S
Steven Rostedt 已提交
9 10 11
#include <linux/spinlock.h>
#include <linux/debugfs.h>
#include <linux/uaccess.h>
12
#include <linux/hardirq.h>
13
#include <linux/kmemcheck.h>
S
Steven Rostedt 已提交
14 15 16 17 18 19
#include <linux/module.h>
#include <linux/percpu.h>
#include <linux/mutex.h>
#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 24
#include "trace.h"

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

32 33
	ret = trace_seq_printf(s, "# compressed entry header\n");
	ret = trace_seq_printf(s, "\ttype_len    :    5 bits\n");
34 35 36 37 38 39 40
	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);
41 42
	ret = trace_seq_printf(s, "\tdata max type_len  == %d\n",
			       RINGBUF_TYPE_DATA_TYPE_LEN_MAX);
43 44 45 46

	return ret;
}

47 48 49 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
/*
 * 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.
 *
 */

115 116 117 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
/*
 * 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,
};

154
static unsigned long ring_buffer_flags __read_mostly = RB_BUFFERS_ON;
155

156 157
#define BUF_PAGE_HDR_SIZE offsetof(struct buffer_data_page, data)

158 159 160 161 162 163 164 165
/**
 * tracing_on - enable all tracing buffers
 *
 * This function enables all tracing buffers that may have been
 * disabled with tracing_off.
 */
void tracing_on(void)
{
166
	set_bit(RB_BUFFERS_ON_BIT, &ring_buffer_flags);
167
}
168
EXPORT_SYMBOL_GPL(tracing_on);
169 170 171 172 173 174 175 176 177 178 179

/**
 * tracing_off - turn off all tracing buffers
 *
 * This function stops all tracing buffers from recording data.
 * It does not disable any overhead the tracers themselves may
 * be causing. This function simply causes all recording to
 * the ring buffers to fail.
 */
void tracing_off(void)
{
180 181
	clear_bit(RB_BUFFERS_ON_BIT, &ring_buffer_flags);
}
182
EXPORT_SYMBOL_GPL(tracing_off);
183 184 185 186 187

/**
 * tracing_off_permanent - permanently disable ring buffers
 *
 * This function, once called, will disable all ring buffers
W
Wenji Huang 已提交
188
 * permanently.
189 190 191 192
 */
void tracing_off_permanent(void)
{
	set_bit(RB_BUFFERS_DISABLED_BIT, &ring_buffer_flags);
193 194
}

195 196 197 198 199 200 201 202 203
/**
 * tracing_is_on - show state of ring buffers enabled
 */
int tracing_is_on(void)
{
	return ring_buffer_flags == RB_BUFFERS_ON;
}
EXPORT_SYMBOL_GPL(tracing_is_on);

204
#define RB_EVNT_HDR_SIZE (offsetof(struct ring_buffer_event, array))
205
#define RB_ALIGNMENT		4U
206
#define RB_MAX_SMALL_DATA	(RB_ALIGNMENT * RINGBUF_TYPE_DATA_TYPE_LEN_MAX)
207
#define RB_EVNT_MIN_SIZE	8U	/* two 32bit words */
208 209 210

/* define RINGBUF_TYPE_DATA for 'case RINGBUF_TYPE_DATA:' */
#define RINGBUF_TYPE_DATA 0 ... RINGBUF_TYPE_DATA_TYPE_LEN_MAX
S
Steven Rostedt 已提交
211 212 213 214 215 216

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

217 218
static inline int rb_null_event(struct ring_buffer_event *event)
{
219
	return event->type_len == RINGBUF_TYPE_PADDING && !event->time_delta;
220 221 222 223
}

static void rb_event_set_padding(struct ring_buffer_event *event)
{
224
	/* padding has a NULL time_delta */
225
	event->type_len = RINGBUF_TYPE_PADDING;
226 227 228
	event->time_delta = 0;
}

229
static unsigned
230
rb_event_data_length(struct ring_buffer_event *event)
S
Steven Rostedt 已提交
231 232 233
{
	unsigned length;

234 235
	if (event->type_len)
		length = event->type_len * RB_ALIGNMENT;
236 237 238 239 240 241 242 243 244
	else
		length = event->array[0];
	return length + RB_EVNT_HDR_SIZE;
}

/* inline for ring buffer fast paths */
static unsigned
rb_event_length(struct ring_buffer_event *event)
{
245
	switch (event->type_len) {
S
Steven Rostedt 已提交
246
	case RINGBUF_TYPE_PADDING:
247 248 249
		if (rb_null_event(event))
			/* undefined */
			return -1;
250
		return  event->array[0] + RB_EVNT_HDR_SIZE;
S
Steven Rostedt 已提交
251 252 253 254 255 256 257 258

	case RINGBUF_TYPE_TIME_EXTEND:
		return RB_LEN_TIME_EXTEND;

	case RINGBUF_TYPE_TIME_STAMP:
		return RB_LEN_TIME_STAMP;

	case RINGBUF_TYPE_DATA:
259
		return rb_event_data_length(event);
S
Steven Rostedt 已提交
260 261 262 263 264 265 266 267 268 269 270 271 272
	default:
		BUG();
	}
	/* not hit */
	return 0;
}

/**
 * ring_buffer_event_length - return the length of the event
 * @event: the event to get the length of
 */
unsigned ring_buffer_event_length(struct ring_buffer_event *event)
{
273
	unsigned length = rb_event_length(event);
274
	if (event->type_len > RINGBUF_TYPE_DATA_TYPE_LEN_MAX)
275 276 277 278 279
		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 已提交
280
}
281
EXPORT_SYMBOL_GPL(ring_buffer_event_length);
S
Steven Rostedt 已提交
282 283

/* inline for ring buffer fast paths */
284
static void *
S
Steven Rostedt 已提交
285 286
rb_event_data(struct ring_buffer_event *event)
{
287
	BUG_ON(event->type_len > RINGBUF_TYPE_DATA_TYPE_LEN_MAX);
S
Steven Rostedt 已提交
288
	/* If length is in len field, then array[0] has the data */
289
	if (event->type_len)
S
Steven Rostedt 已提交
290 291 292 293 294 295 296 297 298 299 300 301 302
		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);
}
303
EXPORT_SYMBOL_GPL(ring_buffer_event_data);
S
Steven Rostedt 已提交
304 305

#define for_each_buffer_cpu(buffer, cpu)		\
306
	for_each_cpu(cpu, buffer->cpumask)
S
Steven Rostedt 已提交
307 308 309 310 311

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

312
struct buffer_data_page {
313
	u64		 time_stamp;	/* page time stamp */
W
Wenji Huang 已提交
314
	local_t		 commit;	/* write committed index */
315 316 317
	unsigned char	 data[];	/* data of buffer page */
};

S
Steven Rostedt 已提交
318 319 320 321 322 323 324 325
/*
 * 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.
 */
326
struct buffer_page {
327
	struct list_head list;		/* list of buffer pages */
328
	local_t		 write;		/* index for next write */
329
	unsigned	 read;		/* index for next read */
330
	local_t		 entries;	/* entries on this page */
331
	struct buffer_data_page *page;	/* Actual data page */
S
Steven Rostedt 已提交
332 333
};

S
Steven Rostedt 已提交
334 335 336 337 338 339 340 341 342 343 344 345 346 347 348
/*
 * 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)

349
static void rb_init_page(struct buffer_data_page *bpage)
350
{
351
	local_set(&bpage->commit, 0);
352 353
}

354 355 356 357 358 359
/**
 * 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.
 */
360 361
size_t ring_buffer_page_len(void *page)
{
362 363
	return local_read(&((struct buffer_data_page *)page)->commit)
		+ BUF_PAGE_HDR_SIZE;
364 365
}

366 367 368 369
/*
 * Also stolen from mm/slob.c. Thanks to Mathieu Desnoyers for pointing
 * this issue out.
 */
370
static void free_buffer_page(struct buffer_page *bpage)
371
{
372
	free_page((unsigned long)bpage->page);
373
	kfree(bpage);
374 375
}

S
Steven Rostedt 已提交
376 377 378 379 380 381 382 383 384 385
/*
 * 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;
}

386
#define BUF_PAGE_SIZE (PAGE_SIZE - BUF_PAGE_HDR_SIZE)
S
Steven Rostedt 已提交
387

388 389 390
/* Max payload is BUF_PAGE_SIZE - header (8bytes) */
#define BUF_MAX_DATA_SIZE (BUF_PAGE_SIZE - (sizeof(u32) * 2))

391 392 393
/* Max number of timestamps that can fit on a page */
#define RB_TIMESTAMPS_PER_PAGE	(BUF_PAGE_SIZE / RB_LEN_TIME_STAMP)

394 395 396 397 398 399
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"
400 401 402
			       "offset:0;\tsize:%u;\tsigned:%u;\n",
			       (unsigned int)sizeof(field.time_stamp),
			       (unsigned int)is_signed_type(u64));
403 404

	ret = trace_seq_printf(s, "\tfield: local_t commit;\t"
405
			       "offset:%u;\tsize:%u;\tsigned:%u;\n",
406
			       (unsigned int)offsetof(typeof(field), commit),
407 408
			       (unsigned int)sizeof(field.commit),
			       (unsigned int)is_signed_type(long));
409 410

	ret = trace_seq_printf(s, "\tfield: char data;\t"
411
			       "offset:%u;\tsize:%u;\tsigned:%u;\n",
412
			       (unsigned int)offsetof(typeof(field), data),
413 414
			       (unsigned int)BUF_PAGE_SIZE,
			       (unsigned int)is_signed_type(char));
415 416 417 418

	return ret;
}

S
Steven Rostedt 已提交
419 420 421 422 423 424
/*
 * head_page == tail_page && head == tail then buffer is empty.
 */
struct ring_buffer_per_cpu {
	int				cpu;
	struct ring_buffer		*buffer;
S
Steven Rostedt 已提交
425
	spinlock_t			reader_lock;	/* serialize readers */
426
	arch_spinlock_t			lock;
S
Steven Rostedt 已提交
427
	struct lock_class_key		lock_key;
428
	struct list_head		*pages;
429 430
	struct buffer_page		*head_page;	/* read from head */
	struct buffer_page		*tail_page;	/* write to tail */
W
Wenji Huang 已提交
431
	struct buffer_page		*commit_page;	/* committed pages */
432
	struct buffer_page		*reader_page;
S
Steven Rostedt 已提交
433 434
	local_t				commit_overrun;
	local_t				overrun;
435
	local_t				entries;
436 437
	local_t				committing;
	local_t				commits;
S
Steven Rostedt 已提交
438
	unsigned long			read;
S
Steven Rostedt 已提交
439 440 441 442 443 444 445 446 447 448
	u64				write_stamp;
	u64				read_stamp;
	atomic_t			record_disabled;
};

struct ring_buffer {
	unsigned			pages;
	unsigned			flags;
	int				cpus;
	atomic_t			record_disabled;
449
	cpumask_var_t			cpumask;
S
Steven Rostedt 已提交
450

451 452
	struct lock_class_key		*reader_lock_key;

S
Steven Rostedt 已提交
453 454 455
	struct mutex			mutex;

	struct ring_buffer_per_cpu	**buffers;
456

457
#ifdef CONFIG_HOTPLUG_CPU
458 459
	struct notifier_block		cpu_notify;
#endif
460
	u64				(*clock)(void);
S
Steven Rostedt 已提交
461 462 463 464 465 466 467 468 469
};

struct ring_buffer_iter {
	struct ring_buffer_per_cpu	*cpu_buffer;
	unsigned long			head;
	struct buffer_page		*head_page;
	u64				read_stamp;
};

470
/* buffer may be either ring_buffer or ring_buffer_per_cpu */
471 472 473 474 475 476 477 478 479 480 481 482 483
#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 已提交
484
	})
485

486 487 488
/* Up this if you want to test the TIME_EXTENTS and normalization */
#define DEBUG_SHIFT 0

489
static inline u64 rb_time_stamp(struct ring_buffer *buffer)
490 491 492 493 494
{
	/* shift to debug/test normalization and TIME_EXTENTS */
	return buffer->clock() << DEBUG_SHIFT;
}

495 496 497 498 499
u64 ring_buffer_time_stamp(struct ring_buffer *buffer, int cpu)
{
	u64 time;

	preempt_disable_notrace();
500
	time = rb_time_stamp(buffer);
501 502 503 504 505 506 507 508 509 510 511 512 513 514
	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 已提交
515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 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
/*
 * 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);
}

/*
605
 * rb_is_head_page - test if the given page is the head page
S
Steven Rostedt 已提交
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 642 643 644 645 646 647 648 649 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
 *
 * 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.
 */
static int inline
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;

705 706
	ret = cmpxchg((unsigned long *)&list->next,
		      val | old_flag, val | new_flag);
S
Steven Rostedt 已提交
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 742 743 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

	/* 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;

798
	ret = cmpxchg(ptr, val, (unsigned long)&new->list);
S
Steven Rostedt 已提交
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 835 836 837 838 839 840 841 842 843 844 845 846 847 848

	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.
849 850 851 852 853
		 *
		 * 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 已提交
854
		 */
855 856
		(void)local_cmpxchg(&next_page->write, old_write, val);
		(void)local_cmpxchg(&next_page->entries, old_entries, eval);
S
Steven Rostedt 已提交
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 886 887 888 889 890 891 892 893 894 895 896 897 898

		/*
		 * 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 已提交
899 900 901 902
/**
 * check_pages - integrity check of buffer pages
 * @cpu_buffer: CPU buffer with pages to test
 *
W
Wenji Huang 已提交
903
 * As a safety measure we check to make sure the data pages have not
S
Steven Rostedt 已提交
904 905 906 907
 * been corrupted.
 */
static int rb_check_pages(struct ring_buffer_per_cpu *cpu_buffer)
{
908
	struct list_head *head = cpu_buffer->pages;
909
	struct buffer_page *bpage, *tmp;
S
Steven Rostedt 已提交
910

S
Steven Rostedt 已提交
911 912
	rb_head_page_deactivate(cpu_buffer);

S
Steven Rostedt 已提交
913 914 915 916
	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 已提交
917

S
Steven Rostedt 已提交
918 919 920
	if (rb_check_list(cpu_buffer, head))
		return -1;

921
	list_for_each_entry_safe(bpage, tmp, head, list) {
S
Steven Rostedt 已提交
922
		if (RB_WARN_ON(cpu_buffer,
923
			       bpage->list.next->prev != &bpage->list))
S
Steven Rostedt 已提交
924 925
			return -1;
		if (RB_WARN_ON(cpu_buffer,
926
			       bpage->list.prev->next != &bpage->list))
S
Steven Rostedt 已提交
927
			return -1;
S
Steven Rostedt 已提交
928 929
		if (rb_check_list(cpu_buffer, &bpage->list))
			return -1;
S
Steven Rostedt 已提交
930 931
	}

S
Steven Rostedt 已提交
932 933
	rb_head_page_activate(cpu_buffer);

S
Steven Rostedt 已提交
934 935 936 937 938 939
	return 0;
}

static int rb_allocate_pages(struct ring_buffer_per_cpu *cpu_buffer,
			     unsigned nr_pages)
{
940
	struct buffer_page *bpage, *tmp;
S
Steven Rostedt 已提交
941 942 943 944
	unsigned long addr;
	LIST_HEAD(pages);
	unsigned i;

945 946
	WARN_ON(!nr_pages);

S
Steven Rostedt 已提交
947
	for (i = 0; i < nr_pages; i++) {
948
		bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),
S
Steven Rostedt 已提交
949
				    GFP_KERNEL, cpu_to_node(cpu_buffer->cpu));
950
		if (!bpage)
951
			goto free_pages;
S
Steven Rostedt 已提交
952 953 954

		rb_check_bpage(cpu_buffer, bpage);

955
		list_add(&bpage->list, &pages);
956

S
Steven Rostedt 已提交
957 958 959
		addr = __get_free_page(GFP_KERNEL);
		if (!addr)
			goto free_pages;
960 961
		bpage->page = (void *)addr;
		rb_init_page(bpage->page);
S
Steven Rostedt 已提交
962 963
	}

964 965 966 967 968 969 970
	/*
	 * 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 已提交
971 972 973 974 975 976

	rb_check_pages(cpu_buffer);

	return 0;

 free_pages:
977 978 979
	list_for_each_entry_safe(bpage, tmp, &pages, list) {
		list_del_init(&bpage->list);
		free_buffer_page(bpage);
S
Steven Rostedt 已提交
980 981 982 983 984 985 986 987
	}
	return -ENOMEM;
}

static struct ring_buffer_per_cpu *
rb_allocate_cpu_buffer(struct ring_buffer *buffer, int cpu)
{
	struct ring_buffer_per_cpu *cpu_buffer;
988
	struct buffer_page *bpage;
989
	unsigned long addr;
S
Steven Rostedt 已提交
990 991 992 993 994 995 996 997 998
	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;
S
Steven Rostedt 已提交
999
	spin_lock_init(&cpu_buffer->reader_lock);
1000
	lockdep_set_class(&cpu_buffer->reader_lock, buffer->reader_lock_key);
1001
	cpu_buffer->lock = (arch_spinlock_t)__ARCH_SPIN_LOCK_UNLOCKED;
S
Steven Rostedt 已提交
1002

1003
	bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),
1004
			    GFP_KERNEL, cpu_to_node(cpu));
1005
	if (!bpage)
1006 1007
		goto fail_free_buffer;

S
Steven Rostedt 已提交
1008 1009
	rb_check_bpage(cpu_buffer, bpage);

1010
	cpu_buffer->reader_page = bpage;
1011 1012
	addr = __get_free_page(GFP_KERNEL);
	if (!addr)
1013
		goto fail_free_reader;
1014 1015
	bpage->page = (void *)addr;
	rb_init_page(bpage->page);
1016

1017 1018
	INIT_LIST_HEAD(&cpu_buffer->reader_page->list);

S
Steven Rostedt 已提交
1019 1020
	ret = rb_allocate_pages(cpu_buffer, buffer->pages);
	if (ret < 0)
1021
		goto fail_free_reader;
S
Steven Rostedt 已提交
1022 1023

	cpu_buffer->head_page
1024
		= list_entry(cpu_buffer->pages, struct buffer_page, list);
S
Steven Rostedt 已提交
1025
	cpu_buffer->tail_page = cpu_buffer->commit_page = cpu_buffer->head_page;
S
Steven Rostedt 已提交
1026

S
Steven Rostedt 已提交
1027 1028
	rb_head_page_activate(cpu_buffer);

S
Steven Rostedt 已提交
1029 1030
	return cpu_buffer;

1031 1032 1033
 fail_free_reader:
	free_buffer_page(cpu_buffer->reader_page);

S
Steven Rostedt 已提交
1034 1035 1036 1037 1038 1039 1040
 fail_free_buffer:
	kfree(cpu_buffer);
	return NULL;
}

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

1044 1045
	free_buffer_page(cpu_buffer->reader_page);

S
Steven Rostedt 已提交
1046 1047
	rb_head_page_deactivate(cpu_buffer);

1048 1049 1050 1051 1052 1053
	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);
1054
		free_buffer_page(bpage);
S
Steven Rostedt 已提交
1055
	}
1056

S
Steven Rostedt 已提交
1057 1058 1059
	kfree(cpu_buffer);
}

1060
#ifdef CONFIG_HOTPLUG_CPU
1061 1062
static int rb_cpu_notify(struct notifier_block *self,
			 unsigned long action, void *hcpu);
1063 1064
#endif

S
Steven Rostedt 已提交
1065 1066
/**
 * ring_buffer_alloc - allocate a new ring_buffer
1067
 * @size: the size in bytes per cpu that is needed.
S
Steven Rostedt 已提交
1068 1069 1070 1071 1072 1073 1074
 * @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.
 */
1075 1076
struct ring_buffer *__ring_buffer_alloc(unsigned long size, unsigned flags,
					struct lock_class_key *key)
S
Steven Rostedt 已提交
1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087
{
	struct ring_buffer *buffer;
	int bsize;
	int cpu;

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

1088 1089 1090
	if (!alloc_cpumask_var(&buffer->cpumask, GFP_KERNEL))
		goto fail_free_buffer;

S
Steven Rostedt 已提交
1091 1092
	buffer->pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
	buffer->flags = flags;
1093
	buffer->clock = trace_clock_local;
1094
	buffer->reader_lock_key = key;
S
Steven Rostedt 已提交
1095 1096

	/* need at least two pages */
1097 1098
	if (buffer->pages < 2)
		buffer->pages = 2;
S
Steven Rostedt 已提交
1099

1100 1101 1102 1103 1104 1105
	/*
	 * 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
1106 1107
	get_online_cpus();
	cpumask_copy(buffer->cpumask, cpu_online_mask);
1108 1109 1110
#else
	cpumask_copy(buffer->cpumask, cpu_possible_mask);
#endif
S
Steven Rostedt 已提交
1111 1112 1113 1114 1115 1116
	buffer->cpus = nr_cpu_ids;

	bsize = sizeof(void *) * nr_cpu_ids;
	buffer->buffers = kzalloc(ALIGN(bsize, cache_line_size()),
				  GFP_KERNEL);
	if (!buffer->buffers)
1117
		goto fail_free_cpumask;
S
Steven Rostedt 已提交
1118 1119 1120 1121 1122 1123 1124 1125

	for_each_buffer_cpu(buffer, cpu) {
		buffer->buffers[cpu] =
			rb_allocate_cpu_buffer(buffer, cpu);
		if (!buffer->buffers[cpu])
			goto fail_free_buffers;
	}

1126
#ifdef CONFIG_HOTPLUG_CPU
1127 1128 1129 1130 1131 1132
	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 已提交
1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143
	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);

1144 1145
 fail_free_cpumask:
	free_cpumask_var(buffer->cpumask);
1146
	put_online_cpus();
1147

S
Steven Rostedt 已提交
1148 1149 1150 1151
 fail_free_buffer:
	kfree(buffer);
	return NULL;
}
1152
EXPORT_SYMBOL_GPL(__ring_buffer_alloc);
S
Steven Rostedt 已提交
1153 1154 1155 1156 1157 1158 1159 1160 1161 1162

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

1163 1164
	get_online_cpus();

1165
#ifdef CONFIG_HOTPLUG_CPU
1166 1167 1168
	unregister_cpu_notifier(&buffer->cpu_notify);
#endif

S
Steven Rostedt 已提交
1169 1170 1171
	for_each_buffer_cpu(buffer, cpu)
		rb_free_cpu_buffer(buffer->buffers[cpu]);

1172 1173
	put_online_cpus();

1174
	kfree(buffer->buffers);
1175 1176
	free_cpumask_var(buffer->cpumask);

S
Steven Rostedt 已提交
1177 1178
	kfree(buffer);
}
1179
EXPORT_SYMBOL_GPL(ring_buffer_free);
S
Steven Rostedt 已提交
1180

1181 1182 1183 1184 1185 1186
void ring_buffer_set_clock(struct ring_buffer *buffer,
			   u64 (*clock)(void))
{
	buffer->clock = clock;
}

S
Steven Rostedt 已提交
1187 1188 1189 1190 1191
static void rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer);

static void
rb_remove_pages(struct ring_buffer_per_cpu *cpu_buffer, unsigned nr_pages)
{
1192
	struct buffer_page *bpage;
S
Steven Rostedt 已提交
1193 1194 1195
	struct list_head *p;
	unsigned i;

1196
	spin_lock_irq(&cpu_buffer->reader_lock);
S
Steven Rostedt 已提交
1197 1198
	rb_head_page_deactivate(cpu_buffer);

S
Steven Rostedt 已提交
1199
	for (i = 0; i < nr_pages; i++) {
1200
		if (RB_WARN_ON(cpu_buffer, list_empty(cpu_buffer->pages)))
S
Steven Rostedt 已提交
1201
			return;
1202
		p = cpu_buffer->pages->next;
1203 1204 1205
		bpage = list_entry(p, struct buffer_page, list);
		list_del_init(&bpage->list);
		free_buffer_page(bpage);
S
Steven Rostedt 已提交
1206
	}
1207
	if (RB_WARN_ON(cpu_buffer, list_empty(cpu_buffer->pages)))
S
Steven Rostedt 已提交
1208
		return;
S
Steven Rostedt 已提交
1209 1210 1211 1212

	rb_reset_cpu(cpu_buffer);
	rb_check_pages(cpu_buffer);

1213
	spin_unlock_irq(&cpu_buffer->reader_lock);
S
Steven Rostedt 已提交
1214 1215 1216 1217 1218 1219
}

static void
rb_insert_pages(struct ring_buffer_per_cpu *cpu_buffer,
		struct list_head *pages, unsigned nr_pages)
{
1220
	struct buffer_page *bpage;
S
Steven Rostedt 已提交
1221 1222 1223
	struct list_head *p;
	unsigned i;

S
Steven Rostedt 已提交
1224 1225 1226
	spin_lock_irq(&cpu_buffer->reader_lock);
	rb_head_page_deactivate(cpu_buffer);

S
Steven Rostedt 已提交
1227
	for (i = 0; i < nr_pages; i++) {
S
Steven Rostedt 已提交
1228 1229
		if (RB_WARN_ON(cpu_buffer, list_empty(pages)))
			return;
S
Steven Rostedt 已提交
1230
		p = pages->next;
1231 1232
		bpage = list_entry(p, struct buffer_page, list);
		list_del_init(&bpage->list);
1233
		list_add_tail(&bpage->list, cpu_buffer->pages);
S
Steven Rostedt 已提交
1234 1235 1236 1237
	}
	rb_reset_cpu(cpu_buffer);
	rb_check_pages(cpu_buffer);

1238
	spin_unlock_irq(&cpu_buffer->reader_lock);
S
Steven Rostedt 已提交
1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253
}

/**
 * ring_buffer_resize - resize the ring buffer
 * @buffer: the buffer to resize.
 * @size: the new size.
 *
 * Minimum size is 2 * BUF_PAGE_SIZE.
 *
 * Returns -1 on failure.
 */
int ring_buffer_resize(struct ring_buffer *buffer, unsigned long size)
{
	struct ring_buffer_per_cpu *cpu_buffer;
	unsigned nr_pages, rm_pages, new_pages;
1254
	struct buffer_page *bpage, *tmp;
S
Steven Rostedt 已提交
1255 1256 1257 1258 1259
	unsigned long buffer_size;
	unsigned long addr;
	LIST_HEAD(pages);
	int i, cpu;

1260 1261 1262 1263 1264 1265
	/*
	 * Always succeed at resizing a non-existent buffer:
	 */
	if (!buffer)
		return size;

S
Steven Rostedt 已提交
1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276
	size = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
	size *= BUF_PAGE_SIZE;
	buffer_size = buffer->pages * BUF_PAGE_SIZE;

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

	if (size == buffer_size)
		return size;

1277 1278 1279 1280 1281
	atomic_inc(&buffer->record_disabled);

	/* Make sure all writers are done with this buffer. */
	synchronize_sched();

S
Steven Rostedt 已提交
1282
	mutex_lock(&buffer->mutex);
1283
	get_online_cpus();
S
Steven Rostedt 已提交
1284 1285 1286 1287 1288 1289

	nr_pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE);

	if (size < buffer_size) {

		/* easy case, just free pages */
1290 1291
		if (RB_WARN_ON(buffer, nr_pages >= buffer->pages))
			goto out_fail;
S
Steven Rostedt 已提交
1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309

		rm_pages = buffer->pages - nr_pages;

		for_each_buffer_cpu(buffer, cpu) {
			cpu_buffer = buffer->buffers[cpu];
			rb_remove_pages(cpu_buffer, rm_pages);
		}
		goto out;
	}

	/*
	 * This is a bit more difficult. We only want to add pages
	 * when we can allocate enough for all CPUs. We do this
	 * by allocating all the pages and storing them on a local
	 * link list. If we succeed in our allocation, then we
	 * add these pages to the cpu_buffers. Otherwise we just free
	 * them all and return -ENOMEM;
	 */
1310 1311
	if (RB_WARN_ON(buffer, nr_pages <= buffer->pages))
		goto out_fail;
1312

S
Steven Rostedt 已提交
1313 1314 1315 1316
	new_pages = nr_pages - buffer->pages;

	for_each_buffer_cpu(buffer, cpu) {
		for (i = 0; i < new_pages; i++) {
1317
			bpage = kzalloc_node(ALIGN(sizeof(*bpage),
1318 1319
						  cache_line_size()),
					    GFP_KERNEL, cpu_to_node(cpu));
1320
			if (!bpage)
1321
				goto free_pages;
1322
			list_add(&bpage->list, &pages);
S
Steven Rostedt 已提交
1323 1324 1325
			addr = __get_free_page(GFP_KERNEL);
			if (!addr)
				goto free_pages;
1326 1327
			bpage->page = (void *)addr;
			rb_init_page(bpage->page);
S
Steven Rostedt 已提交
1328 1329 1330 1331 1332 1333 1334 1335
		}
	}

	for_each_buffer_cpu(buffer, cpu) {
		cpu_buffer = buffer->buffers[cpu];
		rb_insert_pages(cpu_buffer, &pages, new_pages);
	}

1336 1337
	if (RB_WARN_ON(buffer, !list_empty(&pages)))
		goto out_fail;
S
Steven Rostedt 已提交
1338 1339 1340

 out:
	buffer->pages = nr_pages;
1341
	put_online_cpus();
S
Steven Rostedt 已提交
1342 1343
	mutex_unlock(&buffer->mutex);

1344 1345
	atomic_dec(&buffer->record_disabled);

S
Steven Rostedt 已提交
1346 1347 1348
	return size;

 free_pages:
1349 1350 1351
	list_for_each_entry_safe(bpage, tmp, &pages, list) {
		list_del_init(&bpage->list);
		free_buffer_page(bpage);
S
Steven Rostedt 已提交
1352
	}
1353
	put_online_cpus();
1354
	mutex_unlock(&buffer->mutex);
1355
	atomic_dec(&buffer->record_disabled);
S
Steven Rostedt 已提交
1356
	return -ENOMEM;
1357 1358 1359 1360 1361 1362 1363 1364

	/*
	 * Something went totally wrong, and we are too paranoid
	 * to even clean up the mess.
	 */
 out_fail:
	put_online_cpus();
	mutex_unlock(&buffer->mutex);
1365
	atomic_dec(&buffer->record_disabled);
1366
	return -1;
S
Steven Rostedt 已提交
1367
}
1368
EXPORT_SYMBOL_GPL(ring_buffer_resize);
S
Steven Rostedt 已提交
1369

S
Steven Rostedt 已提交
1370
static inline void *
1371
__rb_data_page_index(struct buffer_data_page *bpage, unsigned index)
S
Steven Rostedt 已提交
1372
{
1373
	return bpage->data + index;
S
Steven Rostedt 已提交
1374 1375
}

1376
static inline void *__rb_page_index(struct buffer_page *bpage, unsigned index)
S
Steven Rostedt 已提交
1377
{
1378
	return bpage->page->data + index;
S
Steven Rostedt 已提交
1379 1380 1381
}

static inline struct ring_buffer_event *
1382
rb_reader_event(struct ring_buffer_per_cpu *cpu_buffer)
S
Steven Rostedt 已提交
1383
{
1384 1385 1386 1387
	return __rb_page_index(cpu_buffer->reader_page,
			       cpu_buffer->reader_page->read);
}

S
Steven Rostedt 已提交
1388 1389 1390
static inline struct ring_buffer_event *
rb_iter_head_event(struct ring_buffer_iter *iter)
{
1391
	return __rb_page_index(iter->head_page, iter->head);
S
Steven Rostedt 已提交
1392 1393
}

S
Steven Rostedt 已提交
1394
static inline unsigned long rb_page_write(struct buffer_page *bpage)
S
Steven Rostedt 已提交
1395
{
S
Steven Rostedt 已提交
1396
	return local_read(&bpage->write) & RB_WRITE_MASK;
S
Steven Rostedt 已提交
1397 1398 1399 1400
}

static inline unsigned rb_page_commit(struct buffer_page *bpage)
{
1401
	return local_read(&bpage->page->commit);
S
Steven Rostedt 已提交
1402 1403
}

S
Steven Rostedt 已提交
1404 1405 1406 1407 1408
static inline unsigned long rb_page_entries(struct buffer_page *bpage)
{
	return local_read(&bpage->entries) & RB_WRITE_MASK;
}

S
Steven Rostedt 已提交
1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425
/* Size is determined by what has been commited */
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;

1426
	return (addr & ~PAGE_MASK) - BUF_PAGE_HDR_SIZE;
S
Steven Rostedt 已提交
1427 1428
}

1429
static inline int
1430 1431
rb_event_is_commit(struct ring_buffer_per_cpu *cpu_buffer,
		   struct ring_buffer_event *event)
S
Steven Rostedt 已提交
1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442
{
	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;
}

1443
static void
S
Steven Rostedt 已提交
1444
rb_set_commit_to_write(struct ring_buffer_per_cpu *cpu_buffer)
S
Steven Rostedt 已提交
1445
{
S
Steven Rostedt 已提交
1446 1447
	unsigned long max_count;

S
Steven Rostedt 已提交
1448 1449 1450 1451 1452 1453 1454 1455
	/*
	 * 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.
	 */
1456
 again:
S
Steven Rostedt 已提交
1457 1458
	max_count = cpu_buffer->buffer->pages * 100;

S
Steven Rostedt 已提交
1459
	while (cpu_buffer->commit_page != cpu_buffer->tail_page) {
S
Steven Rostedt 已提交
1460 1461 1462 1463 1464 1465 1466
		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 已提交
1467
		rb_inc_page(cpu_buffer, &cpu_buffer->commit_page);
1468 1469
		cpu_buffer->write_stamp =
			cpu_buffer->commit_page->page->time_stamp;
S
Steven Rostedt 已提交
1470 1471 1472 1473 1474
		/* 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 已提交
1475 1476 1477 1478 1479 1480

		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 已提交
1481 1482
		barrier();
	}
1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493

	/* 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 已提交
1494 1495
}

1496
static void rb_reset_reader_page(struct ring_buffer_per_cpu *cpu_buffer)
S
Steven Rostedt 已提交
1497
{
1498
	cpu_buffer->read_stamp = cpu_buffer->reader_page->page->time_stamp;
1499
	cpu_buffer->reader_page->read = 0;
1500 1501
}

1502
static void rb_inc_iter(struct ring_buffer_iter *iter)
1503 1504 1505 1506 1507 1508 1509 1510 1511 1512
{
	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 已提交
1513
		iter->head_page = rb_set_head_page(cpu_buffer);
1514 1515 1516
	else
		rb_inc_page(cpu_buffer, &iter->head_page);

1517
	iter->read_stamp = iter->head_page->page->time_stamp;
S
Steven Rostedt 已提交
1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531
	iter->head = 0;
}

/**
 * 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.
 */
1532
static void
S
Steven Rostedt 已提交
1533 1534 1535
rb_update_event(struct ring_buffer_event *event,
			 unsigned type, unsigned length)
{
1536
	event->type_len = type;
S
Steven Rostedt 已提交
1537 1538 1539 1540 1541 1542 1543 1544

	switch (type) {

	case RINGBUF_TYPE_PADDING:
	case RINGBUF_TYPE_TIME_EXTEND:
	case RINGBUF_TYPE_TIME_STAMP:
		break;

1545
	case 0:
S
Steven Rostedt 已提交
1546
		length -= RB_EVNT_HDR_SIZE;
1547
		if (length > RB_MAX_SMALL_DATA)
S
Steven Rostedt 已提交
1548
			event->array[0] = length;
1549 1550
		else
			event->type_len = DIV_ROUND_UP(length, RB_ALIGNMENT);
S
Steven Rostedt 已提交
1551 1552 1553 1554 1555 1556
		break;
	default:
		BUG();
	}
}

S
Steven Rostedt 已提交
1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698 1699 1700 1701 1702 1703 1704 1705 1706 1707 1708 1709 1710 1711 1712 1713
/*
 * 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);

		/*
		 * 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;
}

1714
static unsigned rb_calculate_event_length(unsigned length)
S
Steven Rostedt 已提交
1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730
{
	struct ring_buffer_event event; /* Used only for sizeof array */

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

	if (length > RB_MAX_SMALL_DATA)
		length += sizeof(event.array[0]);

	length += RB_EVNT_HDR_SIZE;
	length = ALIGN(length, RB_ALIGNMENT);

	return length;
}

1731 1732 1733 1734 1735 1736 1737 1738 1739 1740 1741 1742 1743 1744 1745 1746 1747
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) {
		local_sub(length, &tail_page->write);
		return;
	}

	event = __rb_page_index(tail_page, tail);
1748
	kmemcheck_annotate_bitfield(event, bitfield);
1749 1750 1751 1752 1753 1754 1755 1756 1757 1758 1759 1760 1761 1762 1763 1764 1765 1766 1767 1768 1769 1770 1771 1772 1773 1774 1775 1776 1777 1778 1779 1780 1781

	/*
	 * 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);
}
1782

S
Steven Rostedt 已提交
1783
static struct ring_buffer_event *
1784 1785 1786
rb_move_tail(struct ring_buffer_per_cpu *cpu_buffer,
	     unsigned long length, unsigned long tail,
	     struct buffer_page *tail_page, u64 *ts)
S
Steven Rostedt 已提交
1787
{
1788
	struct buffer_page *commit_page = cpu_buffer->commit_page;
S
Steven Rostedt 已提交
1789
	struct ring_buffer *buffer = cpu_buffer->buffer;
S
Steven Rostedt 已提交
1790 1791
	struct buffer_page *next_page;
	int ret;
1792 1793 1794 1795 1796 1797 1798 1799 1800 1801 1802

	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 已提交
1803
		local_inc(&cpu_buffer->commit_overrun);
1804 1805 1806
		goto out_reset;
	}

S
Steven Rostedt 已提交
1807 1808 1809 1810 1811 1812 1813 1814 1815 1816 1817 1818 1819 1820 1821
	/*
	 * 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)) {
1822

S
Steven Rostedt 已提交
1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840 1841 1842 1843 1844 1845 1846 1847 1848 1849 1850 1851 1852 1853 1854 1855 1856 1857 1858 1859
		/*
		 * 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;
			}
1860 1861 1862
		}
	}

S
Steven Rostedt 已提交
1863 1864 1865 1866 1867 1868
	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
		 */
1869
		*ts = rb_time_stamp(buffer);
S
Steven Rostedt 已提交
1870
		next_page->page->time_stamp = *ts;
1871 1872
	}

S
Steven Rostedt 已提交
1873
 out_again:
1874

S
Steven Rostedt 已提交
1875
	rb_reset_tail(cpu_buffer, tail_page, tail, length);
1876 1877 1878 1879

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

1880
 out_reset:
1881
	/* reset write */
1882
	rb_reset_tail(cpu_buffer, tail_page, tail, length);
1883

S
Steven Rostedt 已提交
1884
	return NULL;
S
Steven Rostedt 已提交
1885 1886
}

1887 1888 1889 1890
static struct ring_buffer_event *
__rb_reserve_next(struct ring_buffer_per_cpu *cpu_buffer,
		  unsigned type, unsigned long length, u64 *ts)
{
1891
	struct buffer_page *tail_page;
1892 1893 1894 1895 1896
	struct ring_buffer_event *event;
	unsigned long tail, write;

	tail_page = cpu_buffer->tail_page;
	write = local_add_return(length, &tail_page->write);
S
Steven Rostedt 已提交
1897 1898 1899

	/* set write to only the index of the write */
	write &= RB_WRITE_MASK;
1900 1901 1902 1903 1904
	tail = write - length;

	/* See if we shot pass the end of this buffer page */
	if (write > BUF_PAGE_SIZE)
		return rb_move_tail(cpu_buffer, length, tail,
1905
				    tail_page, ts);
1906 1907 1908 1909

	/* We reserved something on the buffer */

	event = __rb_page_index(tail_page, tail);
1910
	kmemcheck_annotate_bitfield(event, bitfield);
1911 1912 1913 1914 1915 1916 1917
	rb_update_event(event, type, length);

	/* The passed in type is zero for DATA */
	if (likely(!type))
		local_inc(&tail_page->entries);

	/*
1918 1919
	 * If this is the first commit on the page, then update
	 * its timestamp.
1920
	 */
1921 1922
	if (!tail)
		tail_page->page->time_stamp = *ts;
1923 1924 1925 1926

	return event;
}

1927 1928 1929 1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942 1943
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);
	old_index = new_index + rb_event_length(event);
	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 已提交
1944 1945
		unsigned long write_mask =
			local_read(&bpage->write) & ~RB_WRITE_MASK;
1946 1947 1948 1949 1950 1951
		/*
		 * 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 已提交
1952 1953
		old_index += write_mask;
		new_index += write_mask;
1954 1955 1956 1957 1958 1959 1960 1961 1962
		index = local_cmpxchg(&bpage->write, old_index, new_index);
		if (index == old_index)
			return 1;
	}

	/* could not discard */
	return 0;
}

S
Steven Rostedt 已提交
1963 1964 1965 1966 1967 1968
static int
rb_add_time_stamp(struct ring_buffer_per_cpu *cpu_buffer,
		  u64 *ts, u64 *delta)
{
	struct ring_buffer_event *event;
	static int once;
S
Steven Rostedt 已提交
1969
	int ret;
S
Steven Rostedt 已提交
1970 1971 1972 1973

	if (unlikely(*delta > (1ULL << 59) && !once++)) {
		printk(KERN_WARNING "Delta way too big! %llu"
		       " ts=%llu write stamp = %llu\n",
1974 1975 1976
		       (unsigned long long)*delta,
		       (unsigned long long)*ts,
		       (unsigned long long)cpu_buffer->write_stamp);
S
Steven Rostedt 已提交
1977 1978 1979 1980 1981 1982 1983 1984 1985 1986 1987 1988
		WARN_ON(1);
	}

	/*
	 * The delta is too big, we to add a
	 * new timestamp.
	 */
	event = __rb_reserve_next(cpu_buffer,
				  RINGBUF_TYPE_TIME_EXTEND,
				  RB_LEN_TIME_EXTEND,
				  ts);
	if (!event)
S
Steven Rostedt 已提交
1989
		return -EBUSY;
S
Steven Rostedt 已提交
1990

S
Steven Rostedt 已提交
1991 1992 1993 1994
	if (PTR_ERR(event) == -EAGAIN)
		return -EAGAIN;

	/* Only a commited time event can update the write stamp */
1995
	if (rb_event_is_commit(cpu_buffer, event)) {
S
Steven Rostedt 已提交
1996
		/*
1997 1998 1999
		 * If this is the first on the page, then it was
		 * updated with the page itself. Try to discard it
		 * and if we can't just make it zero.
S
Steven Rostedt 已提交
2000 2001 2002 2003 2004
		 */
		if (rb_event_index(event)) {
			event->time_delta = *delta & TS_MASK;
			event->array[0] = *delta >> TS_SHIFT;
		} else {
2005 2006 2007 2008 2009 2010
			/* try to discard, since we do not need this */
			if (!rb_try_to_discard(cpu_buffer, event)) {
				/* nope, just zero it */
				event->time_delta = 0;
				event->array[0] = 0;
			}
S
Steven Rostedt 已提交
2011
		}
S
Steven Rostedt 已提交
2012
		cpu_buffer->write_stamp = *ts;
S
Steven Rostedt 已提交
2013 2014 2015
		/* let the caller know this was the commit */
		ret = 1;
	} else {
2016 2017 2018 2019 2020 2021
		/* Try to discard the event */
		if (!rb_try_to_discard(cpu_buffer, event)) {
			/* Darn, this is just wasted space */
			event->time_delta = 0;
			event->array[0] = 0;
		}
2022
		ret = 0;
S
Steven Rostedt 已提交
2023 2024
	}

S
Steven Rostedt 已提交
2025 2026 2027
	*delta = 0;

	return ret;
S
Steven Rostedt 已提交
2028 2029
}

2030 2031 2032 2033 2034 2035 2036 2037 2038 2039 2040 2041 2042 2043 2044 2045 2046 2047 2048 2049 2050 2051 2052 2053 2054 2055 2056 2057 2058 2059 2060 2061 2062 2063 2064 2065 2066 2067
static void rb_start_commit(struct ring_buffer_per_cpu *cpu_buffer)
{
	local_inc(&cpu_buffer->committing);
	local_inc(&cpu_buffer->commits);
}

static void rb_end_commit(struct ring_buffer_per_cpu *cpu_buffer)
{
	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 已提交
2068
static struct ring_buffer_event *
2069 2070
rb_reserve_next_event(struct ring_buffer *buffer,
		      struct ring_buffer_per_cpu *cpu_buffer,
2071
		      unsigned long length)
S
Steven Rostedt 已提交
2072 2073
{
	struct ring_buffer_event *event;
2074
	u64 ts, delta = 0;
S
Steven Rostedt 已提交
2075
	int commit = 0;
2076
	int nr_loops = 0;
S
Steven Rostedt 已提交
2077

2078 2079
	rb_start_commit(cpu_buffer);

2080
#ifdef CONFIG_RING_BUFFER_ALLOW_SWAP
2081 2082 2083 2084 2085 2086 2087 2088 2089 2090 2091 2092
	/*
	 * 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;
	}
2093
#endif
2094

2095
	length = rb_calculate_event_length(length);
S
Steven Rostedt 已提交
2096
 again:
2097 2098 2099 2100 2101 2102 2103 2104 2105
	/*
	 * 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 已提交
2106
	if (RB_WARN_ON(cpu_buffer, ++nr_loops > 1000))
2107
		goto out_fail;
2108

2109
	ts = rb_time_stamp(cpu_buffer->buffer);
S
Steven Rostedt 已提交
2110

S
Steven Rostedt 已提交
2111 2112 2113 2114 2115 2116 2117 2118
	/*
	 * Only the first commit can update the timestamp.
	 * Yes there is a race here. If an interrupt comes in
	 * just after the conditional and it traces too, then it
	 * will also check the deltas. More than one timestamp may
	 * also be made. But only the entry that did the actual
	 * commit will be something other than zero.
	 */
2119 2120 2121
	if (likely(cpu_buffer->tail_page == cpu_buffer->commit_page &&
		   rb_page_write(cpu_buffer->tail_page) ==
		   rb_commit_index(cpu_buffer))) {
2122
		u64 diff;
S
Steven Rostedt 已提交
2123

2124
		diff = ts - cpu_buffer->write_stamp;
S
Steven Rostedt 已提交
2125

2126
		/* make sure this diff is calculated here */
S
Steven Rostedt 已提交
2127 2128 2129 2130
		barrier();

		/* Did the write stamp get updated already? */
		if (unlikely(ts < cpu_buffer->write_stamp))
2131
			goto get_event;
S
Steven Rostedt 已提交
2132

2133 2134
		delta = diff;
		if (unlikely(test_time_stamp(delta))) {
S
Steven Rostedt 已提交
2135

S
Steven Rostedt 已提交
2136 2137
			commit = rb_add_time_stamp(cpu_buffer, &ts, &delta);
			if (commit == -EBUSY)
2138
				goto out_fail;
S
Steven Rostedt 已提交
2139 2140 2141 2142 2143

			if (commit == -EAGAIN)
				goto again;

			RB_WARN_ON(cpu_buffer, commit < 0);
S
Steven Rostedt 已提交
2144
		}
2145
	}
S
Steven Rostedt 已提交
2146

2147
 get_event:
2148
	event = __rb_reserve_next(cpu_buffer, 0, length, &ts);
2149
	if (unlikely(PTR_ERR(event) == -EAGAIN))
S
Steven Rostedt 已提交
2150 2151
		goto again;

2152 2153
	if (!event)
		goto out_fail;
S
Steven Rostedt 已提交
2154

2155
	if (!rb_event_is_commit(cpu_buffer, event))
S
Steven Rostedt 已提交
2156 2157 2158 2159 2160
		delta = 0;

	event->time_delta = delta;

	return event;
2161 2162 2163 2164

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

2167 2168
#ifdef CONFIG_TRACING

2169
#define TRACE_RECURSIVE_DEPTH 16
2170 2171 2172

static int trace_recursive_lock(void)
{
2173
	current->trace_recursion++;
2174

2175 2176
	if (likely(current->trace_recursion < TRACE_RECURSIVE_DEPTH))
		return 0;
2177

2178 2179
	/* Disable all tracing before we do anything else */
	tracing_off_permanent();
2180

S
Steven Rostedt 已提交
2181
	printk_once(KERN_WARNING "Tracing recursion: depth[%ld]:"
2182 2183 2184 2185 2186
		    "HC[%lu]:SC[%lu]:NMI[%lu]\n",
		    current->trace_recursion,
		    hardirq_count() >> HARDIRQ_SHIFT,
		    softirq_count() >> SOFTIRQ_SHIFT,
		    in_nmi());
2187

2188 2189
	WARN_ON_ONCE(1);
	return -1;
2190 2191 2192 2193
}

static void trace_recursive_unlock(void)
{
2194
	WARN_ON_ONCE(!current->trace_recursion);
2195

2196
	current->trace_recursion--;
2197 2198
}

2199 2200 2201 2202 2203 2204 2205
#else

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

#endif

S
Steven Rostedt 已提交
2206 2207
static DEFINE_PER_CPU(int, rb_need_resched);

S
Steven Rostedt 已提交
2208 2209 2210 2211 2212 2213 2214 2215 2216 2217 2218 2219 2220 2221 2222 2223
/**
 * 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 *
2224
ring_buffer_lock_reserve(struct ring_buffer *buffer, unsigned long length)
S
Steven Rostedt 已提交
2225 2226 2227
{
	struct ring_buffer_per_cpu *cpu_buffer;
	struct ring_buffer_event *event;
S
Steven Rostedt 已提交
2228
	int cpu, resched;
S
Steven Rostedt 已提交
2229

2230
	if (ring_buffer_flags != RB_BUFFERS_ON)
2231 2232
		return NULL;

S
Steven Rostedt 已提交
2233 2234 2235
	if (atomic_read(&buffer->record_disabled))
		return NULL;

S
Steven Rostedt 已提交
2236
	/* If we are tracing schedule, we don't want to recurse */
2237
	resched = ftrace_preempt_disable();
S
Steven Rostedt 已提交
2238

2239 2240 2241
	if (trace_recursive_lock())
		goto out_nocheck;

S
Steven Rostedt 已提交
2242 2243
	cpu = raw_smp_processor_id();

2244
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
2245
		goto out;
S
Steven Rostedt 已提交
2246 2247 2248 2249

	cpu_buffer = buffer->buffers[cpu];

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

2252
	if (length > BUF_MAX_DATA_SIZE)
S
Steven Rostedt 已提交
2253
		goto out;
S
Steven Rostedt 已提交
2254

2255
	event = rb_reserve_next_event(buffer, cpu_buffer, length);
S
Steven Rostedt 已提交
2256
	if (!event)
2257
		goto out;
S
Steven Rostedt 已提交
2258

S
Steven Rostedt 已提交
2259 2260 2261 2262 2263 2264 2265 2266
	/*
	 * Need to store resched state on this cpu.
	 * Only the first needs to.
	 */

	if (preempt_count() == 1)
		per_cpu(rb_need_resched, cpu) = resched;

S
Steven Rostedt 已提交
2267 2268
	return event;

2269
 out:
2270 2271 2272
	trace_recursive_unlock();

 out_nocheck:
2273
	ftrace_preempt_enable(resched);
S
Steven Rostedt 已提交
2274 2275
	return NULL;
}
2276
EXPORT_SYMBOL_GPL(ring_buffer_lock_reserve);
S
Steven Rostedt 已提交
2277

2278 2279
static void
rb_update_write_stamp(struct ring_buffer_per_cpu *cpu_buffer,
S
Steven Rostedt 已提交
2280 2281
		      struct ring_buffer_event *event)
{
2282 2283 2284 2285 2286 2287
	/*
	 * The event first in the commit queue updates the
	 * time stamp.
	 */
	if (rb_event_is_commit(cpu_buffer, event))
		cpu_buffer->write_stamp += event->time_delta;
2288
}
S
Steven Rostedt 已提交
2289

2290 2291 2292 2293 2294
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);
2295
	rb_end_commit(cpu_buffer);
S
Steven Rostedt 已提交
2296 2297 2298 2299 2300 2301 2302 2303 2304 2305 2306 2307
}

/**
 * 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,
2308
			      struct ring_buffer_event *event)
S
Steven Rostedt 已提交
2309 2310 2311 2312 2313 2314 2315 2316
{
	struct ring_buffer_per_cpu *cpu_buffer;
	int cpu = raw_smp_processor_id();

	cpu_buffer = buffer->buffers[cpu];

	rb_commit(cpu_buffer, event);

2317 2318
	trace_recursive_unlock();

S
Steven Rostedt 已提交
2319 2320 2321
	/*
	 * Only the last preempt count needs to restore preemption.
	 */
2322 2323 2324
	if (preempt_count() == 1)
		ftrace_preempt_enable(per_cpu(rb_need_resched, cpu));
	else
S
Steven Rostedt 已提交
2325
		preempt_enable_no_resched_notrace();
S
Steven Rostedt 已提交
2326 2327 2328

	return 0;
}
2329
EXPORT_SYMBOL_GPL(ring_buffer_unlock_commit);
S
Steven Rostedt 已提交
2330

2331 2332
static inline void rb_event_discard(struct ring_buffer_event *event)
{
2333 2334 2335
	/* 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;
2336 2337 2338 2339 2340
	/* time delta must be non zero */
	if (!event->time_delta)
		event->time_delta = 1;
}

2341 2342 2343 2344 2345 2346 2347 2348 2349 2350 2351 2352 2353 2354 2355 2356 2357 2358 2359 2360 2361 2362 2363 2364 2365 2366 2367 2368 2369 2370 2371 2372 2373 2374 2375 2376 2377 2378 2379 2380
/*
 * 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);
}

2381 2382 2383 2384 2385
/**
 * ring_buffer_commit_discard - discard an event that has not been committed
 * @buffer: the ring buffer
 * @event: non committed event to discard
 *
2386 2387 2388 2389 2390 2391
 * 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
2392 2393 2394 2395 2396 2397 2398 2399 2400 2401 2402 2403 2404 2405 2406
 * 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 */
2407
	rb_event_discard(event);
2408

2409 2410 2411
	cpu = smp_processor_id();
	cpu_buffer = buffer->buffers[cpu];

2412 2413 2414 2415 2416
	/*
	 * This must only be called if the event has not been
	 * committed yet. Thus we can assume that preemption
	 * is still disabled.
	 */
2417
	RB_WARN_ON(buffer, !local_read(&cpu_buffer->committing));
2418

2419
	rb_decrement_entry(cpu_buffer, event);
2420
	if (rb_try_to_discard(cpu_buffer, event))
2421
		goto out;
2422 2423 2424

	/*
	 * The commit is still visible by the reader, so we
2425
	 * must still update the timestamp.
2426
	 */
2427
	rb_update_write_stamp(cpu_buffer, event);
2428
 out:
2429
	rb_end_commit(cpu_buffer);
2430

2431 2432
	trace_recursive_unlock();

2433 2434 2435 2436 2437 2438 2439 2440 2441 2442 2443
	/*
	 * Only the last preempt count needs to restore preemption.
	 */
	if (preempt_count() == 1)
		ftrace_preempt_enable(per_cpu(rb_need_resched, cpu));
	else
		preempt_enable_no_resched_notrace();

}
EXPORT_SYMBOL_GPL(ring_buffer_discard_commit);

S
Steven Rostedt 已提交
2444 2445 2446 2447 2448 2449 2450 2451 2452 2453 2454 2455 2456 2457 2458 2459 2460 2461 2462 2463 2464
/**
 * 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;
S
Steven Rostedt 已提交
2465
	int cpu, resched;
S
Steven Rostedt 已提交
2466

2467
	if (ring_buffer_flags != RB_BUFFERS_ON)
2468 2469
		return -EBUSY;

S
Steven Rostedt 已提交
2470 2471 2472
	if (atomic_read(&buffer->record_disabled))
		return -EBUSY;

2473
	resched = ftrace_preempt_disable();
S
Steven Rostedt 已提交
2474

S
Steven Rostedt 已提交
2475 2476
	cpu = raw_smp_processor_id();

2477
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
2478
		goto out;
S
Steven Rostedt 已提交
2479 2480 2481 2482 2483 2484

	cpu_buffer = buffer->buffers[cpu];

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

2485 2486 2487
	if (length > BUF_MAX_DATA_SIZE)
		goto out;

2488
	event = rb_reserve_next_event(buffer, cpu_buffer, length);
S
Steven Rostedt 已提交
2489 2490 2491 2492 2493 2494 2495 2496 2497 2498 2499
	if (!event)
		goto out;

	body = rb_event_data(event);

	memcpy(body, data, length);

	rb_commit(cpu_buffer, event);

	ret = 0;
 out:
2500
	ftrace_preempt_enable(resched);
S
Steven Rostedt 已提交
2501 2502 2503

	return ret;
}
2504
EXPORT_SYMBOL_GPL(ring_buffer_write);
S
Steven Rostedt 已提交
2505

2506
static int rb_per_cpu_empty(struct ring_buffer_per_cpu *cpu_buffer)
S
Steven Rostedt 已提交
2507 2508
{
	struct buffer_page *reader = cpu_buffer->reader_page;
S
Steven Rostedt 已提交
2509
	struct buffer_page *head = rb_set_head_page(cpu_buffer);
S
Steven Rostedt 已提交
2510 2511
	struct buffer_page *commit = cpu_buffer->commit_page;

S
Steven Rostedt 已提交
2512 2513 2514 2515
	/* In case of error, head will be NULL */
	if (unlikely(!head))
		return 1;

S
Steven Rostedt 已提交
2516 2517 2518 2519 2520 2521
	return reader->read == rb_page_commit(reader) &&
		(commit == reader ||
		 (commit == head &&
		  head->read == rb_page_commit(commit)));
}

S
Steven Rostedt 已提交
2522 2523 2524 2525 2526 2527 2528 2529 2530 2531 2532 2533 2534
/**
 * 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);
}
2535
EXPORT_SYMBOL_GPL(ring_buffer_record_disable);
S
Steven Rostedt 已提交
2536 2537 2538 2539 2540 2541 2542 2543 2544 2545 2546 2547

/**
 * 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
 * to truely enable the writing (much like preempt_disable).
 */
void ring_buffer_record_enable(struct ring_buffer *buffer)
{
	atomic_dec(&buffer->record_disabled);
}
2548
EXPORT_SYMBOL_GPL(ring_buffer_record_enable);
S
Steven Rostedt 已提交
2549 2550 2551 2552 2553 2554 2555 2556 2557 2558 2559 2560 2561 2562 2563

/**
 * 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;

2564
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
2565
		return;
S
Steven Rostedt 已提交
2566 2567 2568 2569

	cpu_buffer = buffer->buffers[cpu];
	atomic_inc(&cpu_buffer->record_disabled);
}
2570
EXPORT_SYMBOL_GPL(ring_buffer_record_disable_cpu);
S
Steven Rostedt 已提交
2571 2572 2573 2574 2575 2576 2577 2578 2579 2580 2581 2582 2583

/**
 * 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
 * to truely enable the writing (much like preempt_disable).
 */
void ring_buffer_record_enable_cpu(struct ring_buffer *buffer, int cpu)
{
	struct ring_buffer_per_cpu *cpu_buffer;

2584
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
2585
		return;
S
Steven Rostedt 已提交
2586 2587 2588 2589

	cpu_buffer = buffer->buffers[cpu];
	atomic_dec(&cpu_buffer->record_disabled);
}
2590
EXPORT_SYMBOL_GPL(ring_buffer_record_enable_cpu);
S
Steven Rostedt 已提交
2591 2592 2593 2594 2595 2596 2597 2598 2599

/**
 * 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;
2600
	unsigned long ret;
S
Steven Rostedt 已提交
2601

2602
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
2603
		return 0;
S
Steven Rostedt 已提交
2604 2605

	cpu_buffer = buffer->buffers[cpu];
S
Steven Rostedt 已提交
2606
	ret = (local_read(&cpu_buffer->entries) - local_read(&cpu_buffer->overrun))
2607
		- cpu_buffer->read;
2608 2609

	return ret;
S
Steven Rostedt 已提交
2610
}
2611
EXPORT_SYMBOL_GPL(ring_buffer_entries_cpu);
S
Steven Rostedt 已提交
2612 2613 2614 2615 2616 2617 2618 2619 2620

/**
 * 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;
2621
	unsigned long ret;
S
Steven Rostedt 已提交
2622

2623
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
2624
		return 0;
S
Steven Rostedt 已提交
2625 2626

	cpu_buffer = buffer->buffers[cpu];
S
Steven Rostedt 已提交
2627
	ret = local_read(&cpu_buffer->overrun);
2628 2629

	return ret;
S
Steven Rostedt 已提交
2630
}
2631
EXPORT_SYMBOL_GPL(ring_buffer_overrun_cpu);
S
Steven Rostedt 已提交
2632

2633 2634 2635 2636 2637 2638 2639 2640 2641 2642 2643 2644 2645 2646 2647
/**
 * 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 已提交
2648
	ret = local_read(&cpu_buffer->commit_overrun);
2649 2650 2651 2652 2653

	return ret;
}
EXPORT_SYMBOL_GPL(ring_buffer_commit_overrun_cpu);

S
Steven Rostedt 已提交
2654 2655 2656 2657 2658 2659 2660 2661 2662 2663 2664 2665 2666 2667 2668 2669
/**
 * 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];
2670
		entries += (local_read(&cpu_buffer->entries) -
S
Steven Rostedt 已提交
2671
			    local_read(&cpu_buffer->overrun)) - cpu_buffer->read;
S
Steven Rostedt 已提交
2672 2673 2674 2675
	}

	return entries;
}
2676
EXPORT_SYMBOL_GPL(ring_buffer_entries);
S
Steven Rostedt 已提交
2677 2678

/**
2679
 * ring_buffer_overruns - get the number of overruns in buffer
S
Steven Rostedt 已提交
2680 2681 2682 2683 2684 2685 2686 2687 2688 2689 2690 2691 2692 2693
 * @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 已提交
2694
		overruns += local_read(&cpu_buffer->overrun);
S
Steven Rostedt 已提交
2695 2696 2697 2698
	}

	return overruns;
}
2699
EXPORT_SYMBOL_GPL(ring_buffer_overruns);
S
Steven Rostedt 已提交
2700

2701
static void rb_iter_reset(struct ring_buffer_iter *iter)
S
Steven Rostedt 已提交
2702 2703 2704
{
	struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;

2705 2706
	/* Iterator usage is expected to have record disabled */
	if (list_empty(&cpu_buffer->reader_page->list)) {
S
Steven Rostedt 已提交
2707 2708 2709 2710
		iter->head_page = rb_set_head_page(cpu_buffer);
		if (unlikely(!iter->head_page))
			return;
		iter->head = iter->head_page->read;
2711 2712
	} else {
		iter->head_page = cpu_buffer->reader_page;
2713
		iter->head = cpu_buffer->reader_page->read;
2714 2715 2716 2717
	}
	if (iter->head)
		iter->read_stamp = cpu_buffer->read_stamp;
	else
2718
		iter->read_stamp = iter->head_page->page->time_stamp;
2719
}
S
Steven Rostedt 已提交
2720

2721 2722 2723 2724 2725 2726 2727 2728 2729
/**
 * 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)
{
2730
	struct ring_buffer_per_cpu *cpu_buffer;
2731 2732
	unsigned long flags;

2733 2734 2735 2736 2737
	if (!iter)
		return;

	cpu_buffer = iter->cpu_buffer;

2738 2739
	spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
	rb_iter_reset(iter);
S
Steven Rostedt 已提交
2740
	spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
S
Steven Rostedt 已提交
2741
}
2742
EXPORT_SYMBOL_GPL(ring_buffer_iter_reset);
S
Steven Rostedt 已提交
2743 2744 2745 2746 2747 2748 2749 2750 2751 2752 2753

/**
 * 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 已提交
2754 2755
	return iter->head_page == cpu_buffer->commit_page &&
		iter->head == rb_commit_index(cpu_buffer);
S
Steven Rostedt 已提交
2756
}
2757
EXPORT_SYMBOL_GPL(ring_buffer_iter_empty);
S
Steven Rostedt 已提交
2758 2759 2760 2761 2762 2763 2764

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

2765
	switch (event->type_len) {
S
Steven Rostedt 已提交
2766 2767 2768 2769 2770 2771 2772 2773 2774 2775 2776 2777 2778 2779 2780 2781 2782 2783 2784 2785 2786 2787 2788 2789 2790 2791 2792 2793 2794 2795
	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;

2796
	switch (event->type_len) {
S
Steven Rostedt 已提交
2797 2798 2799 2800 2801 2802 2803 2804 2805 2806 2807 2808 2809 2810 2811 2812 2813 2814 2815 2816 2817 2818 2819 2820
	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;
}

2821 2822
static struct buffer_page *
rb_get_reader_page(struct ring_buffer_per_cpu *cpu_buffer)
S
Steven Rostedt 已提交
2823
{
2824 2825
	struct buffer_page *reader = NULL;
	unsigned long flags;
2826
	int nr_loops = 0;
S
Steven Rostedt 已提交
2827
	int ret;
2828

2829
	local_irq_save(flags);
2830
	arch_spin_lock(&cpu_buffer->lock);
2831 2832

 again:
2833 2834 2835 2836 2837 2838
	/*
	 * 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 已提交
2839
	if (RB_WARN_ON(cpu_buffer, ++nr_loops > 3)) {
2840 2841 2842 2843
		reader = NULL;
		goto out;
	}

2844 2845 2846
	reader = cpu_buffer->reader_page;

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

	/* Never should we have an index greater than the size */
S
Steven Rostedt 已提交
2851 2852 2853
	if (RB_WARN_ON(cpu_buffer,
		       cpu_buffer->reader_page->read > rb_page_size(reader)))
		goto out;
2854 2855 2856

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

	/*
2861
	 * Reset the reader page to size zero.
S
Steven Rostedt 已提交
2862
	 */
S
Steven Rostedt 已提交
2863 2864 2865
	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);
S
Steven Rostedt 已提交
2866

S
Steven Rostedt 已提交
2867 2868 2869 2870 2871
 spin:
	/*
	 * Splice the empty reader page into the list around the head.
	 */
	reader = rb_set_head_page(cpu_buffer);
2872 2873
	cpu_buffer->reader_page->list.next = reader->list.next;
	cpu_buffer->reader_page->list.prev = reader->list.prev;
S
Steven Rostedt 已提交
2874

2875 2876 2877 2878 2879 2880 2881
	/*
	 * cpu_buffer->pages just needs to point to the buffer, it
	 *  has no specific buffer page to point to. Lets move it out
	 *  of our way so we don't accidently swap it.
	 */
	cpu_buffer->pages = reader->list.prev;

S
Steven Rostedt 已提交
2882 2883
	/* 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 已提交
2884

S
Steven Rostedt 已提交
2885 2886 2887 2888 2889 2890 2891 2892 2893 2894 2895 2896
	/*
	 * 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 已提交
2897 2898

	/*
S
Steven Rostedt 已提交
2899
	 * If we did not convert it, then we must try again.
S
Steven Rostedt 已提交
2900
	 */
S
Steven Rostedt 已提交
2901 2902
	if (!ret)
		goto spin;
S
Steven Rostedt 已提交
2903

S
Steven Rostedt 已提交
2904 2905 2906 2907 2908 2909 2910
	/*
	 * Yeah! We succeeded in replacing the page.
	 *
	 * Now make the new head point back to the reader page.
	 */
	reader->list.next->prev = &cpu_buffer->reader_page->list;
	rb_inc_page(cpu_buffer, &cpu_buffer->head_page);
2911 2912 2913 2914 2915 2916 2917 2918

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

	goto again;

 out:
2919
	arch_spin_unlock(&cpu_buffer->lock);
2920
	local_irq_restore(flags);
2921 2922 2923 2924 2925 2926 2927 2928 2929 2930 2931

	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 已提交
2932

2933
	/* This function should not be called when buffer is empty */
S
Steven Rostedt 已提交
2934 2935
	if (RB_WARN_ON(cpu_buffer, !reader))
		return;
S
Steven Rostedt 已提交
2936

2937 2938
	event = rb_reader_event(cpu_buffer);

2939
	if (event->type_len <= RINGBUF_TYPE_DATA_TYPE_LEN_MAX)
2940
		cpu_buffer->read++;
2941 2942 2943 2944

	rb_update_read_stamp(cpu_buffer, event);

	length = rb_event_length(event);
2945
	cpu_buffer->reader_page->read += length;
S
Steven Rostedt 已提交
2946 2947 2948 2949 2950 2951 2952 2953 2954 2955 2956 2957 2958 2959 2960
}

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

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

	/*
	 * Check if we are at the end of the buffer.
	 */
S
Steven Rostedt 已提交
2961
	if (iter->head >= rb_page_size(iter->head_page)) {
2962 2963
		/* discarded commits can make the page empty */
		if (iter->head_page == cpu_buffer->commit_page)
S
Steven Rostedt 已提交
2964
			return;
2965
		rb_inc_iter(iter);
S
Steven Rostedt 已提交
2966 2967 2968 2969 2970 2971 2972 2973 2974 2975 2976
		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 已提交
2977
	if (RB_WARN_ON(cpu_buffer,
2978
		       (iter->head_page == cpu_buffer->commit_page) &&
S
Steven Rostedt 已提交
2979 2980
		       (iter->head + length > rb_commit_index(cpu_buffer))))
		return;
S
Steven Rostedt 已提交
2981 2982 2983 2984 2985 2986

	rb_update_iter_read_stamp(iter, event);

	iter->head += length;

	/* check for end of page padding */
S
Steven Rostedt 已提交
2987 2988
	if ((iter->head >= rb_page_size(iter->head_page)) &&
	    (iter->head_page != cpu_buffer->commit_page))
S
Steven Rostedt 已提交
2989 2990 2991
		rb_advance_iter(iter);
}

S
Steven Rostedt 已提交
2992
static struct ring_buffer_event *
2993
rb_buffer_peek(struct ring_buffer_per_cpu *cpu_buffer, u64 *ts)
S
Steven Rostedt 已提交
2994 2995
{
	struct ring_buffer_event *event;
2996
	struct buffer_page *reader;
2997
	int nr_loops = 0;
S
Steven Rostedt 已提交
2998 2999

 again:
3000 3001 3002
	/*
	 * We repeat when a timestamp is encountered. It is possible
	 * to get multiple timestamps from an interrupt entering just
3003 3004
	 * as one timestamp is about to be written, or from discarded
	 * commits. The most that we can have is the number on a single page.
3005
	 */
3006
	if (RB_WARN_ON(cpu_buffer, ++nr_loops > RB_TIMESTAMPS_PER_PAGE))
3007 3008
		return NULL;

3009 3010
	reader = rb_get_reader_page(cpu_buffer);
	if (!reader)
S
Steven Rostedt 已提交
3011 3012
		return NULL;

3013
	event = rb_reader_event(cpu_buffer);
S
Steven Rostedt 已提交
3014

3015
	switch (event->type_len) {
S
Steven Rostedt 已提交
3016
	case RINGBUF_TYPE_PADDING:
3017 3018 3019 3020 3021 3022 3023 3024 3025 3026 3027
		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 已提交
3028 3029 3030

	case RINGBUF_TYPE_TIME_EXTEND:
		/* Internal data, OK to advance */
3031
		rb_advance_reader(cpu_buffer);
S
Steven Rostedt 已提交
3032 3033 3034 3035
		goto again;

	case RINGBUF_TYPE_TIME_STAMP:
		/* FIXME: not implemented */
3036
		rb_advance_reader(cpu_buffer);
S
Steven Rostedt 已提交
3037 3038 3039 3040 3041
		goto again;

	case RINGBUF_TYPE_DATA:
		if (ts) {
			*ts = cpu_buffer->read_stamp + event->time_delta;
3042
			ring_buffer_normalize_time_stamp(cpu_buffer->buffer,
3043
							 cpu_buffer->cpu, ts);
S
Steven Rostedt 已提交
3044 3045 3046 3047 3048 3049 3050 3051 3052
		}
		return event;

	default:
		BUG();
	}

	return NULL;
}
3053
EXPORT_SYMBOL_GPL(ring_buffer_peek);
S
Steven Rostedt 已提交
3054

S
Steven Rostedt 已提交
3055 3056
static struct ring_buffer_event *
rb_iter_peek(struct ring_buffer_iter *iter, u64 *ts)
S
Steven Rostedt 已提交
3057 3058 3059 3060
{
	struct ring_buffer *buffer;
	struct ring_buffer_per_cpu *cpu_buffer;
	struct ring_buffer_event *event;
3061
	int nr_loops = 0;
S
Steven Rostedt 已提交
3062 3063 3064 3065 3066 3067 3068 3069

	if (ring_buffer_iter_empty(iter))
		return NULL;

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

 again:
3070
	/*
3071 3072 3073 3074 3075 3076
	 * We repeat when a timestamp is encountered.
	 * We can get multiple timestamps by nested interrupts or also
	 * if filtering is on (discarding commits). Since discarding
	 * commits can be frequent we can get a lot of timestamps.
	 * But we limit them by not adding timestamps if they begin
	 * at the start of a page.
3077
	 */
3078
	if (RB_WARN_ON(cpu_buffer, ++nr_loops > RB_TIMESTAMPS_PER_PAGE))
3079 3080
		return NULL;

S
Steven Rostedt 已提交
3081 3082 3083 3084 3085
	if (rb_per_cpu_empty(cpu_buffer))
		return NULL;

	event = rb_iter_head_event(iter);

3086
	switch (event->type_len) {
S
Steven Rostedt 已提交
3087
	case RINGBUF_TYPE_PADDING:
3088 3089 3090 3091 3092 3093
		if (rb_null_event(event)) {
			rb_inc_iter(iter);
			goto again;
		}
		rb_advance_iter(iter);
		return event;
S
Steven Rostedt 已提交
3094 3095 3096 3097 3098 3099 3100 3101 3102 3103 3104 3105 3106 3107

	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;
3108 3109
			ring_buffer_normalize_time_stamp(buffer,
							 cpu_buffer->cpu, ts);
S
Steven Rostedt 已提交
3110 3111 3112 3113 3114 3115 3116 3117 3118
		}
		return event;

	default:
		BUG();
	}

	return NULL;
}
3119
EXPORT_SYMBOL_GPL(ring_buffer_iter_peek);
S
Steven Rostedt 已提交
3120

3121 3122 3123 3124 3125 3126 3127 3128
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.
	 */
3129
	if (likely(!in_nmi()))
3130 3131 3132 3133 3134 3135
		return 1;

	tracing_off_permanent();
	return 0;
}

S
Steven Rostedt 已提交
3136 3137 3138 3139 3140 3141 3142 3143 3144 3145 3146 3147 3148
/**
 * 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.
 *
 * This will return the event that will be read next, but does
 * not consume the data.
 */
struct ring_buffer_event *
ring_buffer_peek(struct ring_buffer *buffer, int cpu, u64 *ts)
{
	struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
3149
	struct ring_buffer_event *event;
S
Steven Rostedt 已提交
3150
	unsigned long flags;
3151
	int dolock;
S
Steven Rostedt 已提交
3152

3153
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
3154
		return NULL;
3155

3156
	dolock = rb_ok_to_lock();
3157
 again:
3158 3159 3160
	local_irq_save(flags);
	if (dolock)
		spin_lock(&cpu_buffer->reader_lock);
3161
	event = rb_buffer_peek(cpu_buffer, ts);
3162 3163
	if (event && event->type_len == RINGBUF_TYPE_PADDING)
		rb_advance_reader(cpu_buffer);
3164 3165 3166
	if (dolock)
		spin_unlock(&cpu_buffer->reader_lock);
	local_irq_restore(flags);
S
Steven Rostedt 已提交
3167

3168
	if (event && event->type_len == RINGBUF_TYPE_PADDING)
3169 3170
		goto again;

S
Steven Rostedt 已提交
3171 3172 3173 3174 3175 3176 3177 3178 3179 3180 3181 3182 3183 3184 3185 3186 3187 3188
	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;

3189
 again:
S
Steven Rostedt 已提交
3190 3191 3192 3193
	spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
	event = rb_iter_peek(iter, ts);
	spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);

3194
	if (event && event->type_len == RINGBUF_TYPE_PADDING)
3195 3196
		goto again;

S
Steven Rostedt 已提交
3197 3198 3199
	return event;
}

S
Steven Rostedt 已提交
3200 3201 3202 3203 3204 3205 3206 3207 3208 3209 3210
/**
 * ring_buffer_consume - return an event and consume it
 * @buffer: The ring buffer to get the next event from
 *
 * 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 *
ring_buffer_consume(struct ring_buffer *buffer, int cpu, u64 *ts)
{
3211 3212
	struct ring_buffer_per_cpu *cpu_buffer;
	struct ring_buffer_event *event = NULL;
S
Steven Rostedt 已提交
3213
	unsigned long flags;
3214 3215 3216
	int dolock;

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

3218
 again:
3219 3220 3221
	/* might be called in atomic */
	preempt_disable();

3222
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
3223
		goto out;
S
Steven Rostedt 已提交
3224

3225
	cpu_buffer = buffer->buffers[cpu];
3226 3227 3228
	local_irq_save(flags);
	if (dolock)
		spin_lock(&cpu_buffer->reader_lock);
S
Steven Rostedt 已提交
3229

3230
	event = rb_buffer_peek(cpu_buffer, ts);
3231 3232
	if (event)
		rb_advance_reader(cpu_buffer);
S
Steven Rostedt 已提交
3233

3234 3235 3236
	if (dolock)
		spin_unlock(&cpu_buffer->reader_lock);
	local_irq_restore(flags);
S
Steven Rostedt 已提交
3237

3238 3239 3240
 out:
	preempt_enable();

3241
	if (event && event->type_len == RINGBUF_TYPE_PADDING)
3242 3243
		goto again;

S
Steven Rostedt 已提交
3244 3245
	return event;
}
3246
EXPORT_SYMBOL_GPL(ring_buffer_consume);
S
Steven Rostedt 已提交
3247 3248 3249 3250 3251 3252 3253 3254 3255 3256 3257 3258 3259 3260 3261 3262 3263

/**
 * ring_buffer_read_start - start a non consuming read of the buffer
 * @buffer: The ring buffer to read from
 * @cpu: The cpu buffer to iterate over
 *
 * This starts up an iteration through the buffer. It also disables
 * the recording to the buffer until the reading is finished.
 * This prevents the reading from being corrupted. This is not
 * a consuming read, so a producer is not expected.
 *
 * Must be paired with ring_buffer_finish.
 */
struct ring_buffer_iter *
ring_buffer_read_start(struct ring_buffer *buffer, int cpu)
{
	struct ring_buffer_per_cpu *cpu_buffer;
3264
	struct ring_buffer_iter *iter;
3265
	unsigned long flags;
S
Steven Rostedt 已提交
3266

3267
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
3268
		return NULL;
S
Steven Rostedt 已提交
3269 3270 3271

	iter = kmalloc(sizeof(*iter), GFP_KERNEL);
	if (!iter)
3272
		return NULL;
S
Steven Rostedt 已提交
3273 3274 3275 3276 3277 3278 3279 3280

	cpu_buffer = buffer->buffers[cpu];

	iter->cpu_buffer = cpu_buffer;

	atomic_inc(&cpu_buffer->record_disabled);
	synchronize_sched();

S
Steven Rostedt 已提交
3281
	spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
3282
	arch_spin_lock(&cpu_buffer->lock);
3283
	rb_iter_reset(iter);
3284
	arch_spin_unlock(&cpu_buffer->lock);
S
Steven Rostedt 已提交
3285
	spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
S
Steven Rostedt 已提交
3286 3287 3288

	return iter;
}
3289
EXPORT_SYMBOL_GPL(ring_buffer_read_start);
S
Steven Rostedt 已提交
3290 3291 3292 3293 3294 3295 3296 3297 3298 3299 3300 3301 3302 3303 3304 3305

/**
 * 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;

	atomic_dec(&cpu_buffer->record_disabled);
	kfree(iter);
}
3306
EXPORT_SYMBOL_GPL(ring_buffer_read_finish);
S
Steven Rostedt 已提交
3307 3308 3309 3310 3311 3312 3313 3314 3315 3316 3317 3318

/**
 * 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 已提交
3319 3320
	struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
	unsigned long flags;
S
Steven Rostedt 已提交
3321

S
Steven Rostedt 已提交
3322
	spin_lock_irqsave(&cpu_buffer->reader_lock, flags);
3323
 again:
S
Steven Rostedt 已提交
3324
	event = rb_iter_peek(iter, ts);
S
Steven Rostedt 已提交
3325
	if (!event)
S
Steven Rostedt 已提交
3326
		goto out;
S
Steven Rostedt 已提交
3327

3328 3329 3330
	if (event->type_len == RINGBUF_TYPE_PADDING)
		goto again;

S
Steven Rostedt 已提交
3331
	rb_advance_iter(iter);
S
Steven Rostedt 已提交
3332 3333
 out:
	spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
S
Steven Rostedt 已提交
3334 3335 3336

	return event;
}
3337
EXPORT_SYMBOL_GPL(ring_buffer_read);
S
Steven Rostedt 已提交
3338 3339 3340 3341 3342 3343 3344 3345 3346

/**
 * ring_buffer_size - return the size of the ring buffer (in bytes)
 * @buffer: The ring buffer.
 */
unsigned long ring_buffer_size(struct ring_buffer *buffer)
{
	return BUF_PAGE_SIZE * buffer->pages;
}
3347
EXPORT_SYMBOL_GPL(ring_buffer_size);
S
Steven Rostedt 已提交
3348 3349 3350 3351

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

S
Steven Rostedt 已提交
3354
	cpu_buffer->head_page
3355
		= list_entry(cpu_buffer->pages, struct buffer_page, list);
S
Steven Rostedt 已提交
3356
	local_set(&cpu_buffer->head_page->write, 0);
3357
	local_set(&cpu_buffer->head_page->entries, 0);
3358
	local_set(&cpu_buffer->head_page->page->commit, 0);
3359

3360
	cpu_buffer->head_page->read = 0;
S
Steven Rostedt 已提交
3361 3362 3363 3364 3365 3366

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

	INIT_LIST_HEAD(&cpu_buffer->reader_page->list);
	local_set(&cpu_buffer->reader_page->write, 0);
3367
	local_set(&cpu_buffer->reader_page->entries, 0);
3368
	local_set(&cpu_buffer->reader_page->page->commit, 0);
3369
	cpu_buffer->reader_page->read = 0;
S
Steven Rostedt 已提交
3370

S
Steven Rostedt 已提交
3371 3372
	local_set(&cpu_buffer->commit_overrun, 0);
	local_set(&cpu_buffer->overrun, 0);
3373
	local_set(&cpu_buffer->entries, 0);
3374 3375
	local_set(&cpu_buffer->committing, 0);
	local_set(&cpu_buffer->commits, 0);
S
Steven Rostedt 已提交
3376
	cpu_buffer->read = 0;
3377 3378 3379

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

	rb_head_page_activate(cpu_buffer);
S
Steven Rostedt 已提交
3382 3383 3384 3385 3386 3387 3388 3389 3390 3391 3392 3393
}

/**
 * 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;

3394
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
3395
		return;
S
Steven Rostedt 已提交
3396

3397 3398
	atomic_inc(&cpu_buffer->record_disabled);

S
Steven Rostedt 已提交
3399 3400
	spin_lock_irqsave(&cpu_buffer->reader_lock, flags);

3401 3402 3403
	if (RB_WARN_ON(cpu_buffer, local_read(&cpu_buffer->committing)))
		goto out;

3404
	arch_spin_lock(&cpu_buffer->lock);
S
Steven Rostedt 已提交
3405 3406 3407

	rb_reset_cpu(cpu_buffer);

3408
	arch_spin_unlock(&cpu_buffer->lock);
S
Steven Rostedt 已提交
3409

3410
 out:
S
Steven Rostedt 已提交
3411
	spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
3412 3413

	atomic_dec(&cpu_buffer->record_disabled);
S
Steven Rostedt 已提交
3414
}
3415
EXPORT_SYMBOL_GPL(ring_buffer_reset_cpu);
S
Steven Rostedt 已提交
3416 3417 3418 3419 3420 3421 3422 3423 3424 3425

/**
 * 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)
3426
		ring_buffer_reset_cpu(buffer, cpu);
S
Steven Rostedt 已提交
3427
}
3428
EXPORT_SYMBOL_GPL(ring_buffer_reset);
S
Steven Rostedt 已提交
3429 3430 3431 3432 3433 3434 3435 3436

/**
 * 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;
3437
	unsigned long flags;
3438
	int dolock;
S
Steven Rostedt 已提交
3439
	int cpu;
3440
	int ret;
S
Steven Rostedt 已提交
3441

3442
	dolock = rb_ok_to_lock();
S
Steven Rostedt 已提交
3443 3444 3445 3446

	/* 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];
3447 3448 3449
		local_irq_save(flags);
		if (dolock)
			spin_lock(&cpu_buffer->reader_lock);
3450
		ret = rb_per_cpu_empty(cpu_buffer);
3451 3452 3453 3454
		if (dolock)
			spin_unlock(&cpu_buffer->reader_lock);
		local_irq_restore(flags);

3455
		if (!ret)
S
Steven Rostedt 已提交
3456 3457
			return 0;
	}
3458

S
Steven Rostedt 已提交
3459 3460
	return 1;
}
3461
EXPORT_SYMBOL_GPL(ring_buffer_empty);
S
Steven Rostedt 已提交
3462 3463 3464 3465 3466 3467 3468 3469 3470

/**
 * 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;
3471
	unsigned long flags;
3472
	int dolock;
3473
	int ret;
S
Steven Rostedt 已提交
3474

3475
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
3476
		return 1;
S
Steven Rostedt 已提交
3477

3478 3479
	dolock = rb_ok_to_lock();

S
Steven Rostedt 已提交
3480
	cpu_buffer = buffer->buffers[cpu];
3481 3482 3483
	local_irq_save(flags);
	if (dolock)
		spin_lock(&cpu_buffer->reader_lock);
3484
	ret = rb_per_cpu_empty(cpu_buffer);
3485 3486 3487
	if (dolock)
		spin_unlock(&cpu_buffer->reader_lock);
	local_irq_restore(flags);
3488 3489

	return ret;
S
Steven Rostedt 已提交
3490
}
3491
EXPORT_SYMBOL_GPL(ring_buffer_empty_cpu);
S
Steven Rostedt 已提交
3492

3493
#ifdef CONFIG_RING_BUFFER_ALLOW_SWAP
S
Steven Rostedt 已提交
3494 3495 3496 3497 3498 3499 3500 3501 3502 3503 3504 3505 3506 3507 3508
/**
 * 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;
3509 3510
	int ret = -EINVAL;

3511 3512
	if (!cpumask_test_cpu(cpu, buffer_a->cpumask) ||
	    !cpumask_test_cpu(cpu, buffer_b->cpumask))
3513
		goto out;
S
Steven Rostedt 已提交
3514 3515

	/* At least make sure the two buffers are somewhat the same */
3516
	if (buffer_a->pages != buffer_b->pages)
3517 3518 3519
		goto out;

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

3521
	if (ring_buffer_flags != RB_BUFFERS_ON)
3522
		goto out;
3523 3524

	if (atomic_read(&buffer_a->record_disabled))
3525
		goto out;
3526 3527

	if (atomic_read(&buffer_b->record_disabled))
3528
		goto out;
3529

S
Steven Rostedt 已提交
3530 3531 3532
	cpu_buffer_a = buffer_a->buffers[cpu];
	cpu_buffer_b = buffer_b->buffers[cpu];

3533
	if (atomic_read(&cpu_buffer_a->record_disabled))
3534
		goto out;
3535 3536

	if (atomic_read(&cpu_buffer_b->record_disabled))
3537
		goto out;
3538

S
Steven Rostedt 已提交
3539 3540 3541 3542 3543 3544 3545 3546 3547
	/*
	 * 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);

3548 3549 3550 3551 3552 3553
	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 已提交
3554 3555 3556 3557 3558 3559
	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;

3560 3561 3562
	ret = 0;

out_dec:
S
Steven Rostedt 已提交
3563 3564
	atomic_dec(&cpu_buffer_a->record_disabled);
	atomic_dec(&cpu_buffer_b->record_disabled);
3565 3566
out:
	return ret;
S
Steven Rostedt 已提交
3567
}
3568
EXPORT_SYMBOL_GPL(ring_buffer_swap_cpu);
3569
#endif /* CONFIG_RING_BUFFER_ALLOW_SWAP */
S
Steven Rostedt 已提交
3570

S
Steven Rostedt 已提交
3571 3572 3573 3574 3575 3576 3577 3578 3579 3580 3581 3582 3583 3584 3585 3586 3587
/**
 * 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.
 */
void *ring_buffer_alloc_read_page(struct ring_buffer *buffer)
{
3588
	struct buffer_data_page *bpage;
3589
	unsigned long addr;
S
Steven Rostedt 已提交
3590 3591 3592 3593 3594

	addr = __get_free_page(GFP_KERNEL);
	if (!addr)
		return NULL;

3595
	bpage = (void *)addr;
S
Steven Rostedt 已提交
3596

3597 3598
	rb_init_page(bpage);

3599
	return bpage;
S
Steven Rostedt 已提交
3600
}
S
Steven Rostedt 已提交
3601
EXPORT_SYMBOL_GPL(ring_buffer_alloc_read_page);
S
Steven Rostedt 已提交
3602 3603 3604 3605 3606 3607 3608 3609 3610 3611 3612 3613

/**
 * 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 已提交
3614
EXPORT_SYMBOL_GPL(ring_buffer_free_read_page);
S
Steven Rostedt 已提交
3615 3616 3617 3618 3619

/**
 * 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
3620
 * @len: amount to extract
S
Steven Rostedt 已提交
3621 3622 3623 3624 3625 3626 3627 3628 3629
 * @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 已提交
3630
 *	rpage = ring_buffer_alloc_read_page(buffer);
S
Steven Rostedt 已提交
3631 3632
 *	if (!rpage)
 *		return error;
3633
 *	ret = ring_buffer_read_page(buffer, &rpage, len, cpu, 0);
3634 3635
 *	if (ret >= 0)
 *		process_page(rpage, ret);
S
Steven Rostedt 已提交
3636 3637 3638 3639 3640 3641 3642 3643 3644 3645
 *
 * 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:
3646 3647
 *  >=0 if data has been transferred, returns the offset of consumed data.
 *  <0 if no data has been transferred.
S
Steven Rostedt 已提交
3648 3649
 */
int ring_buffer_read_page(struct ring_buffer *buffer,
3650
			  void **data_page, size_t len, int cpu, int full)
S
Steven Rostedt 已提交
3651 3652 3653
{
	struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];
	struct ring_buffer_event *event;
3654
	struct buffer_data_page *bpage;
3655
	struct buffer_page *reader;
S
Steven Rostedt 已提交
3656
	unsigned long flags;
3657
	unsigned int commit;
3658
	unsigned int read;
3659
	u64 save_timestamp;
3660
	int ret = -1;
S
Steven Rostedt 已提交
3661

3662 3663 3664
	if (!cpumask_test_cpu(cpu, buffer->cpumask))
		goto out;

3665 3666 3667 3668 3669
	/*
	 * If len is not big enough to hold the page header, then
	 * we can not copy anything.
	 */
	if (len <= BUF_PAGE_HDR_SIZE)
3670
		goto out;
3671 3672 3673

	len -= BUF_PAGE_HDR_SIZE;

S
Steven Rostedt 已提交
3674
	if (!data_page)
3675
		goto out;
S
Steven Rostedt 已提交
3676

3677 3678
	bpage = *data_page;
	if (!bpage)
3679
		goto out;
S
Steven Rostedt 已提交
3680 3681 3682

	spin_lock_irqsave(&cpu_buffer->reader_lock, flags);

3683 3684
	reader = rb_get_reader_page(cpu_buffer);
	if (!reader)
3685
		goto out_unlock;
S
Steven Rostedt 已提交
3686

3687 3688 3689 3690
	event = rb_reader_event(cpu_buffer);

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

S
Steven Rostedt 已提交
3692
	/*
3693 3694 3695 3696 3697
	 * 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 已提交
3698
	 */
3699
	if (read || (len < (commit - read)) ||
3700
	    cpu_buffer->reader_page == cpu_buffer->commit_page) {
3701
		struct buffer_data_page *rpage = cpu_buffer->reader_page->page;
3702 3703
		unsigned int rpos = read;
		unsigned int pos = 0;
3704
		unsigned int size;
S
Steven Rostedt 已提交
3705 3706

		if (full)
3707
			goto out_unlock;
S
Steven Rostedt 已提交
3708

3709 3710 3711 3712 3713 3714
		if (len > (commit - read))
			len = (commit - read);

		size = rb_event_length(event);

		if (len < size)
3715
			goto out_unlock;
3716

3717 3718 3719
		/* save the current timestamp, since the user will need it */
		save_timestamp = cpu_buffer->read_stamp;

3720 3721
		/* Need to copy one event at a time */
		do {
3722
			memcpy(bpage->data + pos, rpage->data + rpos, size);
3723 3724 3725 3726

			len -= size;

			rb_advance_reader(cpu_buffer);
3727 3728
			rpos = reader->read;
			pos += size;
3729 3730 3731 3732

			event = rb_reader_event(cpu_buffer);
			size = rb_event_length(event);
		} while (len > size);
3733 3734

		/* update bpage */
3735
		local_set(&bpage->commit, pos);
3736
		bpage->time_stamp = save_timestamp;
3737

3738 3739
		/* we copied everything to the beginning */
		read = 0;
S
Steven Rostedt 已提交
3740
	} else {
3741
		/* update the entry counter */
S
Steven Rostedt 已提交
3742
		cpu_buffer->read += rb_page_entries(reader);
3743

S
Steven Rostedt 已提交
3744
		/* swap the pages */
3745
		rb_init_page(bpage);
3746 3747 3748
		bpage = reader->page;
		reader->page = *data_page;
		local_set(&reader->write, 0);
3749
		local_set(&reader->entries, 0);
3750
		reader->read = 0;
3751
		*data_page = bpage;
S
Steven Rostedt 已提交
3752
	}
3753
	ret = read;
S
Steven Rostedt 已提交
3754

3755
 out_unlock:
S
Steven Rostedt 已提交
3756 3757
	spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);

3758
 out:
S
Steven Rostedt 已提交
3759 3760
	return ret;
}
S
Steven Rostedt 已提交
3761
EXPORT_SYMBOL_GPL(ring_buffer_read_page);
S
Steven Rostedt 已提交
3762

3763
#ifdef CONFIG_TRACING
3764 3765 3766 3767
static ssize_t
rb_simple_read(struct file *filp, char __user *ubuf,
	       size_t cnt, loff_t *ppos)
{
3768
	unsigned long *p = filp->private_data;
3769 3770 3771
	char buf[64];
	int r;

3772 3773 3774 3775
	if (test_bit(RB_BUFFERS_DISABLED_BIT, p))
		r = sprintf(buf, "permanently disabled\n");
	else
		r = sprintf(buf, "%d\n", test_bit(RB_BUFFERS_ON_BIT, p));
3776 3777 3778 3779 3780 3781 3782 3783

	return simple_read_from_buffer(ubuf, cnt, ppos, buf, r);
}

static ssize_t
rb_simple_write(struct file *filp, const char __user *ubuf,
		size_t cnt, loff_t *ppos)
{
3784
	unsigned long *p = filp->private_data;
3785
	char buf[64];
3786
	unsigned long val;
3787 3788 3789 3790 3791 3792 3793 3794 3795 3796 3797 3798 3799 3800
	int ret;

	if (cnt >= sizeof(buf))
		return -EINVAL;

	if (copy_from_user(&buf, ubuf, cnt))
		return -EFAULT;

	buf[cnt] = 0;

	ret = strict_strtoul(buf, 10, &val);
	if (ret < 0)
		return ret;

3801 3802 3803 3804
	if (val)
		set_bit(RB_BUFFERS_ON_BIT, p);
	else
		clear_bit(RB_BUFFERS_ON_BIT, p);
3805 3806 3807 3808 3809 3810

	(*ppos)++;

	return cnt;
}

3811
static const struct file_operations rb_simple_fops = {
3812 3813 3814 3815 3816 3817 3818 3819 3820 3821 3822 3823
	.open		= tracing_open_generic,
	.read		= rb_simple_read,
	.write		= rb_simple_write,
};


static __init int rb_init_debugfs(void)
{
	struct dentry *d_tracer;

	d_tracer = tracing_init_dentry();

3824 3825
	trace_create_file("tracing_on", 0644, d_tracer,
			    &ring_buffer_flags, &rb_simple_fops);
3826 3827 3828 3829 3830

	return 0;
}

fs_initcall(rb_init_debugfs);
3831
#endif
3832

3833
#ifdef CONFIG_HOTPLUG_CPU
3834 3835
static int rb_cpu_notify(struct notifier_block *self,
			 unsigned long action, void *hcpu)
3836 3837 3838 3839 3840 3841 3842 3843
{
	struct ring_buffer *buffer =
		container_of(self, struct ring_buffer, cpu_notify);
	long cpu = (long)hcpu;

	switch (action) {
	case CPU_UP_PREPARE:
	case CPU_UP_PREPARE_FROZEN:
3844
		if (cpumask_test_cpu(cpu, buffer->cpumask))
3845 3846 3847 3848 3849 3850 3851 3852 3853 3854
			return NOTIFY_OK;

		buffer->buffers[cpu] =
			rb_allocate_cpu_buffer(buffer, cpu);
		if (!buffer->buffers[cpu]) {
			WARN(1, "failed to allocate ring buffer on CPU %ld\n",
			     cpu);
			return NOTIFY_OK;
		}
		smp_wmb();
3855
		cpumask_set_cpu(cpu, buffer->cpumask);
3856 3857 3858 3859 3860 3861 3862 3863 3864 3865 3866 3867 3868 3869 3870
		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