ring-buffer: add benchmark and tester

This patch adds code that can benchmark the ring buffer as well as test it. This code can be compiled into the kernel (not recommended) or as a module. A separate ring buffer is used to not interfer with other users, like ftrace. It creates a producer and a consumer (option to disable creation of the consumer) and will run for 10 seconds, then sleep for 10 seconds and then repeat. While running, the producer will write 10 byte loads into the ring buffer with just putting in the current CPU number. The reader will continually try to read the buffer. The reader will alternate from reading the buffer via event by event, or by full pages. The output is a pr_info, thus it will fill up the syslogs. Starting ring buffer hammer End ring buffer hammer Time: 9000349 (usecs) Overruns: 12578640 Read: 5358440 (by events) Entries: 0 Total: 17937080 Missed: 0 Hit: 17937080 Entries per millisec: 1993 501 ns per entry Sleeping for 10 secs Starting ring buffer hammer End ring buffer hammer Time: 9936350 (usecs) Overruns: 0 Read: 28146644 (by pages) Entries: 74 Total: 28146718 Missed: 0 Hit: 28146718 Entries per millisec: 2832 353 ns per entry Sleeping for 10 secs Time: is the time the test ran Overruns: the number of events that were overwritten and not read Read: the number of events read (either by pages or events) Entries: the number of entries left in the buffer (the by pages will only read full pages) Total: Entries + Read + Overruns Missed: the number of entries that failed to write Hit: the number of entries that were written The above example shows that it takes ~353 nanosecs per entry when there is a reader, reading by pages (and no overruns) The event by event reader slowed the producer down to 501 nanosecs. [ Impact: see how changes to the ring buffer affect stability and performance ] Signed-off-by: Steven Rostedt <rostedt@goodmis.org>

ring-buffer: add benchmark and tester
This patch adds code that can benchmark the ring buffer as well as test it. This code can be compiled into the kernel (not recommended) or as a module. A separate ring buffer is used to not interfer with other users, like ftrace. It creates a producer and a consumer (option to disable creation of the consumer) and will run for 10 seconds, then sleep for 10 seconds and then repeat. While running, the producer will write 10 byte loads into the ring buffer with just putting in the current CPU number. The reader will continually try to read the buffer. The reader will alternate from reading the buffer via event by event, or by full pages. The output is a pr_info, thus it will fill up the syslogs. Starting ring buffer hammer End ring buffer hammer Time: 9000349 (usecs) Overruns: 12578640 Read: 5358440 (by events) Entries: 0 Total: 17937080 Missed: 0 Hit: 17937080 Entries per millisec: 1993 501 ns per entry Sleeping for 10 secs Starting ring buffer hammer End ring buffer hammer Time: 9936350 (usecs) Overruns: 0 Read: 28146644 (by pages) Entries: 74 Total: 28146718 Missed: 0 Hit: 28146718 Entries per millisec: 2832 353 ns per entry Sleeping for 10 secs Time: is the time the test ran Overruns: the number of events that were overwritten and not read Read: the number of events read (either by pages or events) Entries: the number of entries left in the buffer (the by pages will only read full pages) Total: Entries + Read + Overruns Missed: the number of entries that failed to write Hit: the number of entries that were written The above example shows that it takes ~353 nanosecs per entry when there is a reader, reading by pages (and no overruns) The event by event reader slowed the producer down to 501 nanosecs. [ Impact: see how changes to the ring buffer affect stability and performance ] Signed-off-by: Steven Rostedt <rostedt@goodmis.org>
Steven Rostedt · Steven Rostedt
1 parent aa20ae8444
Showing 3 changed files with 396 additions and 0 deletions Side-by-side Diff
kernel/trace/Kconfig
kernel/trace/Makefile
kernel/trace/ring_buffer_benchmark.c
@@ -471,6 +471,22 @@
  
 	  Say N, unless you absolutely know what you are doing.
  
+config RING_BUFFER_BENCHMARK
+	tristate "Ring buffer benchmark stress tester"
+	depends on RING_BUFFER
+	help
+	  This option creates a test to stress the ring buffer and bench mark it.
+	  It creates its own ring buffer such that it will not interfer with
+	  any other users of the ring buffer (such as ftrace). It then creates
+	  a producer and consumer that will run for 10 seconds and sleep for
+	  10 seconds. Each interval it will print out the number of events
+	  it recorded and give a rough estimate of how long each iteration took.
+
+	  It does not disable interrupts or raise its priority, so it may be
+	  affected by processes that are running.
+
+	  If unsure, say N
+
 endif # FTRACE
  
 endif # TRACING_SUPPORT
@@ -17,6 +17,7 @@
  
 obj-$(CONFIG_FUNCTION_TRACER) += libftrace.o
 obj-$(CONFIG_RING_BUFFER) += ring_buffer.o
+obj-$(CONFIG_RING_BUFFER_BENCHMARK) += ring_buffer_benchmark.o
  
 obj-$(CONFIG_TRACING) += trace.o
 obj-$(CONFIG_TRACING) += trace_clock.o
+/*
+ * 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");
...	...	@@ -471,6 +471,22 @@
471	471
472	472	Say N, unless you absolutely know what you are doing.
473	473
	474	+config RING_BUFFER_BENCHMARK
	475	+ tristate "Ring buffer benchmark stress tester"
	476	+ depends on RING_BUFFER
	477	+ help
	478	+ This option creates a test to stress the ring buffer and bench mark it.
	479	+ It creates its own ring buffer such that it will not interfer with
	480	+ any other users of the ring buffer (such as ftrace). It then creates
	481	+ a producer and consumer that will run for 10 seconds and sleep for
	482	+ 10 seconds. Each interval it will print out the number of events
	483	+ it recorded and give a rough estimate of how long each iteration took.
	484	+
	485	+ It does not disable interrupts or raise its priority, so it may be
	486	+ affected by processes that are running.
	487	+
	488	+ If unsure, say N
	489	+
474	490	endif # FTRACE
475	491
476	492	endif # TRACING_SUPPORT
...	...	@@ -17,6 +17,7 @@
17	17
18	18	obj-$(CONFIG_FUNCTION_TRACER) += libftrace.o
19	19	obj-$(CONFIG_RING_BUFFER) += ring_buffer.o
	20	+obj-$(CONFIG_RING_BUFFER_BENCHMARK) += ring_buffer_benchmark.o
20	21
21	22	obj-$(CONFIG_TRACING) += trace.o
22	23	obj-$(CONFIG_TRACING) += trace_clock.o
	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");