[deliverable/linux.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);
	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);
}

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);

	} 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);
87	ret = ring_buffer_read_page(buffer, &bpage, PAGE_SIZE, cpu, 1);
88	if (ret >= 0) {
89	rpage = bpage;
90	commit = local_read(&rpage->commit);
91	for (i = 0; i < commit && !kill_test; i += inc) {
92
93	if (i >= (PAGE_SIZE - offsetof(struct rb_page, data))) {
94	KILL_TEST();
95	break;
96	}
97
98	inc = -1;
99	event = (void *)&rpage->data[i];
100	switch (event->type_len) {
101	case RINGBUF_TYPE_PADDING:
102	/* We don't expect any padding */
103	KILL_TEST();
104	break;
105	case RINGBUF_TYPE_TIME_EXTEND:
106	inc = 8;
107	break;
108	case 0:
109	entry = ring_buffer_event_data(event);
110	if (*entry != cpu) {
111	KILL_TEST();
112	break;
113	}
114	read++;
115	if (!event->array[0]) {
116	KILL_TEST();
117	break;
118	}
119	inc = event->array[0];
120	break;
121	default:
122	entry = ring_buffer_event_data(event);
123	if (*entry != cpu) {
124	KILL_TEST();
125	break;
126	}
127	read++;
128	inc = ((event->type_len + 1) * 4);
129	}
130	if (kill_test)
131	break;
132
133	if (inc <= 0) {
134	KILL_TEST();
135	break;
136	}
137	}
138	}
139	ring_buffer_free_read_page(buffer, bpage);
140
141	if (ret < 0)
142	return EVENT_DROPPED;
143	return EVENT_FOUND;
144	}
145
146	static void ring_buffer_consumer(void)
147	{
148	/* toggle between reading pages and events */
149	read_events ^= 1;
150
151	read = 0;
152	while (!reader_finish && !kill_test) {
153	int found;
154
155	do {
156	int cpu;
157
158	found = 0;
159	for_each_online_cpu(cpu) {
160	enum event_status stat;
161
162	if (read_events)
163	stat = read_event(cpu);
164	else
165	stat = read_page(cpu);
166
167	if (kill_test)
168	break;
169	if (stat == EVENT_FOUND)
170	found = 1;
171	}
172	} while (found && !kill_test);
173
174	set_current_state(TASK_INTERRUPTIBLE);
175	if (reader_finish)
176	break;
177
178	schedule();
179	__set_current_state(TASK_RUNNING);
180	}
181	reader_finish = 0;
182	complete(&read_done);
183	}
184
185	static void ring_buffer_producer(void)
186	{
187	struct timeval start_tv;
188	struct timeval end_tv;
189	unsigned long long time;
190	unsigned long long entries;
191	unsigned long long overruns;
192	unsigned long missed = 0;
193	unsigned long hit = 0;
194	unsigned long avg;
195	int cnt = 0;
196
197	/*
198	* Hammer the buffer for 10 secs (this may
199	* make the system stall)
200	*/
201	pr_info("Starting ring buffer hammer\n");
202	do_gettimeofday(&start_tv);
203	do {
204	struct ring_buffer_event *event;
205	int *entry;
206
207	event = ring_buffer_lock_reserve(buffer, 10);
208	if (!event) {
209	missed++;
210	} else {
211	hit++;
212	entry = ring_buffer_event_data(event);
213	*entry = smp_processor_id();
214	ring_buffer_unlock_commit(buffer, event);
215	}
216	do_gettimeofday(&end_tv);
217
218	if (consumer && !(++cnt % wakeup_interval))
219	wake_up_process(consumer);
220
221	} while (end_tv.tv_sec < (start_tv.tv_sec + RUN_TIME) && !kill_test);
222	pr_info("End ring buffer hammer\n");
223
224	if (consumer) {
225	/* Init both completions here to avoid races */
226	init_completion(&read_start);
227	init_completion(&read_done);
228	/* the completions must be visible before the finish var */
229	smp_wmb();
230	reader_finish = 1;
231	/* finish var visible before waking up the consumer */
232	smp_wmb();
233	wake_up_process(consumer);
234	wait_for_completion(&read_done);
235	}
236
237	time = end_tv.tv_sec - start_tv.tv_sec;
238	time *= 1000000;
239	time += (long long)((long)end_tv.tv_usec - (long)start_tv.tv_usec);
240
241	entries = ring_buffer_entries(buffer);
242	overruns = ring_buffer_overruns(buffer);
243
244	if (kill_test)
245	pr_info("ERROR!\n");
246	pr_info("Time: %lld (usecs)\n", time);
247	pr_info("Overruns: %lld\n", overruns);
248	if (disable_reader)
249	pr_info("Read: (reader disabled)\n");
250	else
251	pr_info("Read: %ld (by %s)\n", read,
252	read_events ? "events" : "pages");
253	pr_info("Entries: %lld\n", entries);
254	pr_info("Total: %lld\n", entries + overruns + read);
255	pr_info("Missed: %ld\n", missed);
256	pr_info("Hit: %ld\n", hit);
257
258	do_div(time, 1000);
259	if (time)
260	hit /= (long)time;
261	else
262	pr_info("TIME IS ZERO??\n");
263
264	pr_info("Entries per millisec: %ld\n", hit);
265
266	if (hit) {
267	avg = 1000000 / hit;
268	pr_info("%ld ns per entry\n", avg);
269	}
270	}
271
272	static void wait_to_die(void)
273	{
274	set_current_state(TASK_INTERRUPTIBLE);
275	while (!kthread_should_stop()) {
276	schedule();
277	set_current_state(TASK_INTERRUPTIBLE);
278	}
279	__set_current_state(TASK_RUNNING);
280	}
281
282	static int ring_buffer_consumer_thread(void *arg)
283	{
284	while (!kthread_should_stop() && !kill_test) {
285	complete(&read_start);
286
287	ring_buffer_consumer();
288
289	set_current_state(TASK_INTERRUPTIBLE);
290	if (kthread_should_stop() \|\| kill_test)
291	break;
292
293	schedule();
294	__set_current_state(TASK_RUNNING);
295	}
296	__set_current_state(TASK_RUNNING);
297
298	if (kill_test)
299	wait_to_die();
300
301	return 0;
302	}
303
304	static int ring_buffer_producer_thread(void *arg)
305	{
306	init_completion(&read_start);
307
308	while (!kthread_should_stop() && !kill_test) {
309	ring_buffer_reset(buffer);
310
311	if (consumer) {
312	smp_wmb();
313	wake_up_process(consumer);
314	wait_for_completion(&read_start);
315	}
316
317	ring_buffer_producer();
318
319	pr_info("Sleeping for 10 secs\n");
320	set_current_state(TASK_INTERRUPTIBLE);
321	schedule_timeout(HZ * SLEEP_TIME);
322	__set_current_state(TASK_RUNNING);
323	}
324
325	if (kill_test)
326	wait_to_die();
327
328	return 0;
329	}
330
331	static int __init ring_buffer_benchmark_init(void)
332	{
333	int ret;
334
335	/* make a one meg buffer in overwite mode */
336	buffer = ring_buffer_alloc(1000000, RB_FL_OVERWRITE);
337	if (!buffer)
338	return -ENOMEM;
339
340	if (!disable_reader) {
341	consumer = kthread_create(ring_buffer_consumer_thread,
342	NULL, "rb_consumer");
343	ret = PTR_ERR(consumer);
344	if (IS_ERR(consumer))
345	goto out_fail;
346	}
347
348	producer = kthread_run(ring_buffer_producer_thread,
349	NULL, "rb_producer");
350	ret = PTR_ERR(producer);
351
352	if (IS_ERR(producer))
353	goto out_kill;
354
355	return 0;
356
357	out_kill:
358	if (consumer)
359	kthread_stop(consumer);
360
361	out_fail:
362	ring_buffer_free(buffer);
363	return ret;
364	}
365
366	static void __exit ring_buffer_benchmark_exit(void)
367	{
368	kthread_stop(producer);
369	if (consumer)
370	kthread_stop(consumer);
371	ring_buffer_free(buffer);
372	}
373
374	module_init(ring_buffer_benchmark_init);
375	module_exit(ring_buffer_benchmark_exit);
376
377	MODULE_AUTHOR("Steven Rostedt");
378	MODULE_DESCRIPTION("ring_buffer_benchmark");
379	MODULE_LICENSE("GPL");