[net-next-2.6.git] / kernel / trace / ring_buffer_benchmark.c

/*
 * ring buffer tester and benchmark
 *
 * Copyright (C) 2009 Steven Rostedt <srostedt@redhat.com>
 */
#include <linux/ring_buffer.h>
#include <linux/completion.h>
#include <linux/kthread.h>
#include <linux/module.h>
#include <linux/time.h>

struct rb_page {
	u64		ts;
	local_t		commit;
	char		data[4080];
};

/* run time and sleep time in seconds */
#define RUN_TIME	10
#define SLEEP_TIME	10

/* number of events for writer to wake up the reader */
static int wakeup_interval = 100;

static int reader_finish;
static struct completion read_start;
static struct completion read_done;

static struct ring_buffer *buffer;
static struct task_struct *producer;
static struct task_struct *consumer;
static unsigned long read;

static int disable_reader;
module_param(disable_reader, uint, 0644);
MODULE_PARM_DESC(disable_reader, "only run producer");

static int read_events;

static int kill_test;

#define KILL_TEST()				\
	do {					\
		if (!kill_test) {		\
			kill_test = 1;		\
			WARN_ON(1);		\
		}				\
	} while (0)

enum event_status {
	EVENT_FOUND,
	EVENT_DROPPED,
};

static enum event_status read_event(int cpu)
{
	struct ring_buffer_event *event;
	int *entry;
	u64 ts;

	event = ring_buffer_consume(buffer, cpu, &ts);
	if (!event)
		return EVENT_DROPPED;

	entry = ring_buffer_event_data(event);
	if (*entry != cpu) {
		KILL_TEST();
		return EVENT_DROPPED;
	}

	read++;
	return EVENT_FOUND;
}

static enum event_status read_page(int cpu)
{
	struct ring_buffer_event *event;
	struct rb_page *rpage;
	unsigned long commit;
	void *bpage;
	int *entry;
	int ret;
	int inc;
	int i;

	bpage = ring_buffer_alloc_read_page(buffer);
	if (!bpage)
		return EVENT_DROPPED;

	ret = ring_buffer_read_page(buffer, &bpage, PAGE_SIZE, cpu, 1);
	if (ret >= 0) {
		rpage = bpage;
		commit = local_read(&rpage->commit);
		for (i = 0; i < commit && !kill_test; i += inc) {

			if (i >= (PAGE_SIZE - offsetof(struct rb_page, data))) {
				KILL_TEST();
				break;
			}

			inc = -1;
			event = (void *)&rpage->data[i];
			switch (event->type_len) {
			case RINGBUF_TYPE_PADDING:
				/* We don't expect any padding */
				KILL_TEST();
				break;
			case RINGBUF_TYPE_TIME_EXTEND:
				inc = 8;
				break;
			case 0:
				entry = ring_buffer_event_data(event);
				if (*entry != cpu) {
					KILL_TEST();
					break;
				}
				read++;
				if (!event->array[0]) {
					KILL_TEST();
					break;
				}
				inc = event->array[0];
				break;
			default:
				entry = ring_buffer_event_data(event);
				if (*entry != cpu) {
					KILL_TEST();
					break;
				}
				read++;
				inc = ((event->type_len + 1) * 4);
			}
			if (kill_test)
				break;

			if (inc <= 0) {
				KILL_TEST();
				break;
			}
		}
	}
	ring_buffer_free_read_page(buffer, bpage);

	if (ret < 0)
		return EVENT_DROPPED;
	return EVENT_FOUND;
}

static void ring_buffer_consumer(void)
{
	/* toggle between reading pages and events */
	read_events ^= 1;

	read = 0;
	while (!reader_finish && !kill_test) {
		int found;

		do {
			int cpu;

			found = 0;
			for_each_online_cpu(cpu) {
				enum event_status stat;

				if (read_events)
					stat = read_event(cpu);
				else
					stat = read_page(cpu);

				if (kill_test)
					break;
				if (stat == EVENT_FOUND)
					found = 1;
			}
		} while (found && !kill_test);

		set_current_state(TASK_INTERRUPTIBLE);
		if (reader_finish)
			break;

		schedule();
		__set_current_state(TASK_RUNNING);
	}
	reader_finish = 0;
	complete(&read_done);
}

/*
 * If we are a non preempt kernel, the 10 second run will
 * stop everything while it runs. Instead, we will call cond_resched
 * and also add any time that was lost by a rescedule.
 */
#ifdef CONFIG_PREEMPT
static void sched_if_needed(struct timeval *start_tv, struct timeval *end_tv)
{
}
#else
static void sched_if_needed(struct timeval *start_tv, struct timeval *end_tv)
{
	struct timeval tv;

	cond_resched();
	do_gettimeofday(&tv);
	if (tv.tv_usec < end_tv->tv_usec) {
		tv.tv_usec += 1000000;
		tv.tv_sec--;
	}
	start_tv->tv_sec += tv.tv_sec - end_tv->tv_sec;
	start_tv->tv_usec += tv.tv_usec - end_tv->tv_usec;
	if (start_tv->tv_usec > 1000000) {
		start_tv->tv_usec -= 1000000;
		start_tv->tv_sec++;
	}
}
#endif

static void ring_buffer_producer(void)
{
	struct timeval start_tv;
	struct timeval end_tv;
	unsigned long long time;
	unsigned long long entries;
	unsigned long long overruns;
	unsigned long missed = 0;
	unsigned long hit = 0;
	unsigned long avg;
	int cnt = 0;

	/*
	 * Hammer the buffer for 10 secs (this may
	 * make the system stall)
	 */
	pr_info("Starting ring buffer hammer\n");
	do_gettimeofday(&start_tv);
	do {
		struct ring_buffer_event *event;
		int *entry;

		event = ring_buffer_lock_reserve(buffer, 10);
		if (!event) {
			missed++;
		} else {
			hit++;
			entry = ring_buffer_event_data(event);
			*entry = smp_processor_id();
			ring_buffer_unlock_commit(buffer, event);
		}
		do_gettimeofday(&end_tv);

		if (consumer && !(++cnt % wakeup_interval))
			wake_up_process(consumer);

		sched_if_needed(&start_tv, &end_tv);

	} while (end_tv.tv_sec < (start_tv.tv_sec + RUN_TIME) && !kill_test);
	pr_info("End ring buffer hammer\n");

	if (consumer) {
		/* Init both completions here to avoid races */
		init_completion(&read_start);
		init_completion(&read_done);
		/* the completions must be visible before the finish var */
		smp_wmb();
		reader_finish = 1;
		/* finish var visible before waking up the consumer */
		smp_wmb();
		wake_up_process(consumer);
		wait_for_completion(&read_done);
	}

	time = end_tv.tv_sec - start_tv.tv_sec;
	time *= 1000000;
	time += (long long)((long)end_tv.tv_usec - (long)start_tv.tv_usec);

	entries = ring_buffer_entries(buffer);
	overruns = ring_buffer_overruns(buffer);

	if (kill_test)
		pr_info("ERROR!\n");
	pr_info("Time:     %lld (usecs)\n", time);
	pr_info("Overruns: %lld\n", overruns);
	if (disable_reader)
		pr_info("Read:     (reader disabled)\n");
	else
		pr_info("Read:     %ld  (by %s)\n", read,
			read_events ? "events" : "pages");
	pr_info("Entries:  %lld\n", entries);
	pr_info("Total:    %lld\n", entries + overruns + read);
	pr_info("Missed:   %ld\n", missed);
	pr_info("Hit:      %ld\n", hit);

	do_div(time, 1000);
	if (time)
		hit /= (long)time;
	else
		pr_info("TIME IS ZERO??\n");

	pr_info("Entries per millisec: %ld\n", hit);

	if (hit) {
		avg = 1000000 / hit;
		pr_info("%ld ns per entry\n", avg);
	}
}

static void wait_to_die(void)
{
	set_current_state(TASK_INTERRUPTIBLE);
	while (!kthread_should_stop()) {
		schedule();
		set_current_state(TASK_INTERRUPTIBLE);
	}
	__set_current_state(TASK_RUNNING);
}

static int ring_buffer_consumer_thread(void *arg)
{
	while (!kthread_should_stop() && !kill_test) {
		complete(&read_start);

		ring_buffer_consumer();

		set_current_state(TASK_INTERRUPTIBLE);
		if (kthread_should_stop() || kill_test)
			break;

		schedule();
		__set_current_state(TASK_RUNNING);
	}
	__set_current_state(TASK_RUNNING);

	if (kill_test)
		wait_to_die();

	return 0;
}

static int ring_buffer_producer_thread(void *arg)
{
	init_completion(&read_start);

	while (!kthread_should_stop() && !kill_test) {
		ring_buffer_reset(buffer);

		if (consumer) {
			smp_wmb();
			wake_up_process(consumer);
			wait_for_completion(&read_start);
		}

		ring_buffer_producer();

		pr_info("Sleeping for 10 secs\n");
		set_current_state(TASK_INTERRUPTIBLE);
		schedule_timeout(HZ * SLEEP_TIME);
		__set_current_state(TASK_RUNNING);
	}

	if (kill_test)
		wait_to_die();

	return 0;
}

static int __init ring_buffer_benchmark_init(void)
{
	int ret;

	/* make a one meg buffer in overwite mode */
	buffer = ring_buffer_alloc(1000000, RB_FL_OVERWRITE);
	if (!buffer)
		return -ENOMEM;

	if (!disable_reader) {
		consumer = kthread_create(ring_buffer_consumer_thread,
					  NULL, "rb_consumer");
		ret = PTR_ERR(consumer);
		if (IS_ERR(consumer))
			goto out_fail;
	}

	producer = kthread_run(ring_buffer_producer_thread,
			       NULL, "rb_producer");
	ret = PTR_ERR(producer);

	if (IS_ERR(producer))
		goto out_kill;

	return 0;

 out_kill:
	if (consumer)
		kthread_stop(consumer);

 out_fail:
	ring_buffer_free(buffer);
	return ret;
}

static void __exit ring_buffer_benchmark_exit(void)
{
	kthread_stop(producer);
	if (consumer)
		kthread_stop(consumer);
	ring_buffer_free(buffer);
}

module_init(ring_buffer_benchmark_init);
module_exit(ring_buffer_benchmark_exit);

MODULE_AUTHOR("Steven Rostedt");
MODULE_DESCRIPTION("ring_buffer_benchmark");
MODULE_LICENSE("GPL");
Commit	Line	Data
5092dbc9 SR	1	/*
	2	* ring buffer tester and benchmark
	3	*
	4	* Copyright (C) 2009 Steven Rostedt <srostedt@redhat.com>
	5	*/
	6	#include <linux/ring_buffer.h>
	7	#include <linux/completion.h>
	8	#include <linux/kthread.h>
	9	#include <linux/module.h>
	10	#include <linux/time.h>
	11
	12	struct rb_page {
	13	u64 ts;
	14	local_t commit;
	15	char data[4080];
	16	};
	17
	18	/* run time and sleep time in seconds */
	19	#define RUN_TIME 10
	20	#define SLEEP_TIME 10
	21
	22	/* number of events for writer to wake up the reader */
	23	static int wakeup_interval = 100;
	24
	25	static int reader_finish;
	26	static struct completion read_start;
	27	static struct completion read_done;
	28
	29	static struct ring_buffer *buffer;
	30	static struct task_struct *producer;
	31	static struct task_struct *consumer;
	32	static unsigned long read;
	33
	34	static int disable_reader;
	35	module_param(disable_reader, uint, 0644);
	36	MODULE_PARM_DESC(disable_reader, "only run producer");
	37
	38	static int read_events;
	39
	40	static int kill_test;
	41
	42	#define KILL_TEST() \
	43	do { \
	44	if (!kill_test) { \
	45	kill_test = 1; \
	46	WARN_ON(1); \
	47	} \
	48	} while (0)
	49
	50	enum event_status {
	51	EVENT_FOUND,
	52	EVENT_DROPPED,
	53	};
	54
	55	static enum event_status read_event(int cpu)
	56	{
	57	struct ring_buffer_event *event;
	58	int *entry;
	59	u64 ts;
	60
	61	event = ring_buffer_consume(buffer, cpu, &ts);
	62	if (!event)
	63	return EVENT_DROPPED;
	64
65	entry = ring_buffer_event_data(event);
66	if (*entry != cpu) {
67	KILL_TEST();
68	return EVENT_DROPPED;
69	}
70
71	read++;
72	return EVENT_FOUND;
73	}
74
75	static enum event_status read_page(int cpu)
76	{
77	struct ring_buffer_event *event;
78	struct rb_page *rpage;
79	unsigned long commit;
80	void *bpage;
81	int *entry;
82	int ret;
83	int inc;
84	int i;
85
86	bpage = ring_buffer_alloc_read_page(buffer);
00c81a58 SR	87	if (!bpage)
	88	return EVENT_DROPPED;
	89
5092dbc9 SR	90	ret = ring_buffer_read_page(buffer, &bpage, PAGE_SIZE, cpu, 1);
	91	if (ret >= 0) {
	92	rpage = bpage;
	93	commit = local_read(&rpage->commit);
	94	for (i = 0; i < commit && !kill_test; i += inc) {
	95
	96	if (i >= (PAGE_SIZE - offsetof(struct rb_page, data))) {
	97	KILL_TEST();
	98	break;
	99	}
	100
	101	inc = -1;
	102	event = (void *)&rpage->data[i];
	103	switch (event->type_len) {
	104	case RINGBUF_TYPE_PADDING:
	105	/* We don't expect any padding */
	106	KILL_TEST();
	107	break;
	108	case RINGBUF_TYPE_TIME_EXTEND:
	109	inc = 8;
	110	break;
	111	case 0:
	112	entry = ring_buffer_event_data(event);
	113	if (*entry != cpu) {
	114	KILL_TEST();
	115	break;
	116	}
	117	read++;
	118	if (!event->array[0]) {
	119	KILL_TEST();
	120	break;
	121	}
	122	inc = event->array[0];
	123	break;
	124	default:
	125	entry = ring_buffer_event_data(event);
	126	if (*entry != cpu) {
	127	KILL_TEST();
	128	break;
	129	}
	130	read++;
	131	inc = ((event->type_len + 1) * 4);
	132	}
	133	if (kill_test)
	134	break;
	135
	136	if (inc <= 0) {
	137	KILL_TEST();
	138	break;
	139	}
	140	}
	141	}
	142	ring_buffer_free_read_page(buffer, bpage);
	143
	144	if (ret < 0)
	145	return EVENT_DROPPED;
	146	return EVENT_FOUND;
	147	}
	148
	149	static void ring_buffer_consumer(void)
	150	{
	151	/* toggle between reading pages and events */
	152	read_events ^= 1;
	153
154	read = 0;
155	while (!reader_finish && !kill_test) {
156	int found;
157
158	do {
159	int cpu;
160
161	found = 0;
162	for_each_online_cpu(cpu) {
163	enum event_status stat;
164
165	if (read_events)
166	stat = read_event(cpu);
167	else
168	stat = read_page(cpu);
169
170	if (kill_test)
171	break;
172	if (stat == EVENT_FOUND)
173	found = 1;
174	}
175	} while (found && !kill_test);
176
177	set_current_state(TASK_INTERRUPTIBLE);
178	if (reader_finish)
179	break;
180
181	schedule();
182	__set_current_state(TASK_RUNNING);
183	}
184	reader_finish = 0;
185	complete(&read_done);
186	}
187
3e07a4f6 SR	188	/*
	189	* If we are a non preempt kernel, the 10 second run will
	190	* stop everything while it runs. Instead, we will call cond_resched
	191	* and also add any time that was lost by a rescedule.
	192	*/
	193	#ifdef CONFIG_PREEMPT
	194	static void sched_if_needed(struct timeval start_tv, struct timeval end_tv)
	195	{
	196	}
	197	#else
	198	static void sched_if_needed(struct timeval start_tv, struct timeval end_tv)
	199	{
	200	struct timeval tv;
	201
	202	cond_resched();
	203	do_gettimeofday(&tv);
	204	if (tv.tv_usec < end_tv->tv_usec) {
	205	tv.tv_usec += 1000000;
	206	tv.tv_sec--;
	207	}
	208	start_tv->tv_sec += tv.tv_sec - end_tv->tv_sec;
	209	start_tv->tv_usec += tv.tv_usec - end_tv->tv_usec;
	210	if (start_tv->tv_usec > 1000000) {
	211	start_tv->tv_usec -= 1000000;
	212	start_tv->tv_sec++;
	213	}
	214	}
	215	#endif
	216
5092dbc9 SR	217	static void ring_buffer_producer(void)
	218	{
	219	struct timeval start_tv;
	220	struct timeval end_tv;
	221	unsigned long long time;
	222	unsigned long long entries;
	223	unsigned long long overruns;
	224	unsigned long missed = 0;
	225	unsigned long hit = 0;
	226	unsigned long avg;
	227	int cnt = 0;
	228
	229	/*
	230	* Hammer the buffer for 10 secs (this may
	231	* make the system stall)
	232	*/
	233	pr_info("Starting ring buffer hammer\n");
	234	do_gettimeofday(&start_tv);
	235	do {
	236	struct ring_buffer_event *event;
	237	int *entry;
	238
	239	event = ring_buffer_lock_reserve(buffer, 10);
	240	if (!event) {
	241	missed++;
	242	} else {
	243	hit++;
	244	entry = ring_buffer_event_data(event);
	245	*entry = smp_processor_id();
	246	ring_buffer_unlock_commit(buffer, event);
	247	}
	248	do_gettimeofday(&end_tv);
	249
	250	if (consumer && !(++cnt % wakeup_interval))
	251	wake_up_process(consumer);
	252
3e07a4f6 SR	253	sched_if_needed(&start_tv, &end_tv);
3e07a4f6 SR	254
5092dbc9 SR	255	} while (end_tv.tv_sec < (start_tv.tv_sec + RUN_TIME) && !kill_test);
	256	pr_info("End ring buffer hammer\n");
	257
	258	if (consumer) {
	259	/* Init both completions here to avoid races */
	260	init_completion(&read_start);
	261	init_completion(&read_done);
	262	/* the completions must be visible before the finish var */
	263	smp_wmb();
	264	reader_finish = 1;
	265	/* finish var visible before waking up the consumer */
	266	smp_wmb();
	267	wake_up_process(consumer);
	268	wait_for_completion(&read_done);
	269	}
	270
	271	time = end_tv.tv_sec - start_tv.tv_sec;
	272	time *= 1000000;
	273	time += (long long)((long)end_tv.tv_usec - (long)start_tv.tv_usec);
	274
	275	entries = ring_buffer_entries(buffer);
	276	overruns = ring_buffer_overruns(buffer);
	277
	278	if (kill_test)
	279	pr_info("ERROR!\n");
	280	pr_info("Time: %lld (usecs)\n", time);
	281	pr_info("Overruns: %lld\n", overruns);
	282	if (disable_reader)
	283	pr_info("Read: (reader disabled)\n");
	284	else
	285	pr_info("Read: %ld (by %s)\n", read,
	286	read_events ? "events" : "pages");
	287	pr_info("Entries: %lld\n", entries);
	288	pr_info("Total: %lld\n", entries + overruns + read);
	289	pr_info("Missed: %ld\n", missed);
	290	pr_info("Hit: %ld\n", hit);
	291
	292	do_div(time, 1000);
	293	if (time)
	294	hit /= (long)time;
	295	else
	296	pr_info("TIME IS ZERO??\n");
	297
	298	pr_info("Entries per millisec: %ld\n", hit);
	299
	300	if (hit) {
	301	avg = 1000000 / hit;
	302	pr_info("%ld ns per entry\n", avg);
	303	}
	304	}
	305
	306	static void wait_to_die(void)
	307	{
	308	set_current_state(TASK_INTERRUPTIBLE);
	309	while (!kthread_should_stop()) {
	310	schedule();
	311	set_current_state(TASK_INTERRUPTIBLE);
	312	}
	313	__set_current_state(TASK_RUNNING);
	314	}
	315
	316	static int ring_buffer_consumer_thread(void *arg)
	317	{
	318	while (!kthread_should_stop() && !kill_test) {
319	complete(&read_start);
320
321	ring_buffer_consumer();
322
323	set_current_state(TASK_INTERRUPTIBLE);
324	if (kthread_should_stop() \|\| kill_test)
325	break;
326
327	schedule();
328	__set_current_state(TASK_RUNNING);
329	}
330	__set_current_state(TASK_RUNNING);
331
332	if (kill_test)
333	wait_to_die();
334
335	return 0;
336	}
337
338	static int ring_buffer_producer_thread(void *arg)
339	{
340	init_completion(&read_start);
341
342	while (!kthread_should_stop() && !kill_test) {
343	ring_buffer_reset(buffer);
344
345	if (consumer) {
346	smp_wmb();
347	wake_up_process(consumer);
348	wait_for_completion(&read_start);
349	}
350
351	ring_buffer_producer();
352
353	pr_info("Sleeping for 10 secs\n");
354	set_current_state(TASK_INTERRUPTIBLE);
355	schedule_timeout(HZ * SLEEP_TIME);
356	__set_current_state(TASK_RUNNING);
357	}
358
359	if (kill_test)
360	wait_to_die();
361
362	return 0;
363	}
364
365	static int __init ring_buffer_benchmark_init(void)
366	{
367	int ret;
368
369	/* make a one meg buffer in overwite mode */
370	buffer = ring_buffer_alloc(1000000, RB_FL_OVERWRITE);
371	if (!buffer)
372	return -ENOMEM;
373
374	if (!disable_reader) {
375	consumer = kthread_create(ring_buffer_consumer_thread,
376	NULL, "rb_consumer");
377	ret = PTR_ERR(consumer);
378	if (IS_ERR(consumer))
379	goto out_fail;
380	}
381
382	producer = kthread_run(ring_buffer_producer_thread,
383	NULL, "rb_producer");
384	ret = PTR_ERR(producer);
385
386	if (IS_ERR(producer))
387	goto out_kill;
388
389	return 0;
390
391	out_kill:
392	if (consumer)
393	kthread_stop(consumer);
394
395	out_fail:
396	ring_buffer_free(buffer);
397	return ret;
398	}
399
400	static void __exit ring_buffer_benchmark_exit(void)
401	{
402	kthread_stop(producer);
403	if (consumer)
404	kthread_stop(consumer);
405	ring_buffer_free(buffer);
406	}
407
408	module_init(ring_buffer_benchmark_init);
409	module_exit(ring_buffer_benchmark_exit);
410
411	MODULE_AUTHOR("Steven Rostedt");
412	MODULE_DESCRIPTION("ring_buffer_benchmark");
413	MODULE_LICENSE("GPL");