ring-buffer: Add place holder recording of dropped events
[linux-2.6.git] / kernel / trace / ring_buffer_benchmark.c
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 #include <asm/local.h>
12
13 struct rb_page {
14         u64             ts;
15         local_t         commit;
16         char            data[4080];
17 };
18
19 /* run time and sleep time in seconds */
20 #define RUN_TIME        10
21 #define SLEEP_TIME      10
22
23 /* number of events for writer to wake up the reader */
24 static int wakeup_interval = 100;
25
26 static int reader_finish;
27 static struct completion read_start;
28 static struct completion read_done;
29
30 static struct ring_buffer *buffer;
31 static struct task_struct *producer;
32 static struct task_struct *consumer;
33 static unsigned long read;
34
35 static int disable_reader;
36 module_param(disable_reader, uint, 0644);
37 MODULE_PARM_DESC(disable_reader, "only run producer");
38
39 static int write_iteration = 50;
40 module_param(write_iteration, uint, 0644);
41 MODULE_PARM_DESC(write_iteration, "# of writes between timestamp readings");
42
43 static int producer_nice = 19;
44 static int consumer_nice = 19;
45
46 static int producer_fifo = -1;
47 static int consumer_fifo = -1;
48
49 module_param(producer_nice, uint, 0644);
50 MODULE_PARM_DESC(producer_nice, "nice prio for producer");
51
52 module_param(consumer_nice, uint, 0644);
53 MODULE_PARM_DESC(consumer_nice, "nice prio for consumer");
54
55 module_param(producer_fifo, uint, 0644);
56 MODULE_PARM_DESC(producer_fifo, "fifo prio for producer");
57
58 module_param(consumer_fifo, uint, 0644);
59 MODULE_PARM_DESC(consumer_fifo, "fifo prio for consumer");
60
61 static int read_events;
62
63 static int kill_test;
64
65 #define KILL_TEST()                             \
66         do {                                    \
67                 if (!kill_test) {               \
68                         kill_test = 1;          \
69                         WARN_ON(1);             \
70                 }                               \
71         } while (0)
72
73 enum event_status {
74         EVENT_FOUND,
75         EVENT_DROPPED,
76 };
77
78 static enum event_status read_event(int cpu)
79 {
80         struct ring_buffer_event *event;
81         int *entry;
82         u64 ts;
83
84         event = ring_buffer_consume(buffer, cpu, &ts, NULL);
85         if (!event)
86                 return EVENT_DROPPED;
87
88         entry = ring_buffer_event_data(event);
89         if (*entry != cpu) {
90                 KILL_TEST();
91                 return EVENT_DROPPED;
92         }
93
94         read++;
95         return EVENT_FOUND;
96 }
97
98 static enum event_status read_page(int cpu)
99 {
100         struct ring_buffer_event *event;
101         struct rb_page *rpage;
102         unsigned long commit;
103         void *bpage;
104         int *entry;
105         int ret;
106         int inc;
107         int i;
108
109         bpage = ring_buffer_alloc_read_page(buffer);
110         if (!bpage)
111                 return EVENT_DROPPED;
112
113         ret = ring_buffer_read_page(buffer, &bpage, PAGE_SIZE, cpu, 1);
114         if (ret >= 0) {
115                 rpage = bpage;
116                 commit = local_read(&rpage->commit);
117                 for (i = 0; i < commit && !kill_test; i += inc) {
118
119                         if (i >= (PAGE_SIZE - offsetof(struct rb_page, data))) {
120                                 KILL_TEST();
121                                 break;
122                         }
123
124                         inc = -1;
125                         event = (void *)&rpage->data[i];
126                         switch (event->type_len) {
127                         case RINGBUF_TYPE_PADDING:
128                                 /* failed writes may be discarded events */
129                                 if (!event->time_delta)
130                                         KILL_TEST();
131                                 inc = event->array[0] + 4;
132                                 break;
133                         case RINGBUF_TYPE_TIME_EXTEND:
134                                 inc = 8;
135                                 break;
136                         case 0:
137                                 entry = ring_buffer_event_data(event);
138                                 if (*entry != cpu) {
139                                         KILL_TEST();
140                                         break;
141                                 }
142                                 read++;
143                                 if (!event->array[0]) {
144                                         KILL_TEST();
145                                         break;
146                                 }
147                                 inc = event->array[0] + 4;
148                                 break;
149                         default:
150                                 entry = ring_buffer_event_data(event);
151                                 if (*entry != cpu) {
152                                         KILL_TEST();
153                                         break;
154                                 }
155                                 read++;
156                                 inc = ((event->type_len + 1) * 4);
157                         }
158                         if (kill_test)
159                                 break;
160
161                         if (inc <= 0) {
162                                 KILL_TEST();
163                                 break;
164                         }
165                 }
166         }
167         ring_buffer_free_read_page(buffer, bpage);
168
169         if (ret < 0)
170                 return EVENT_DROPPED;
171         return EVENT_FOUND;
172 }
173
174 static void ring_buffer_consumer(void)
175 {
176         /* toggle between reading pages and events */
177         read_events ^= 1;
178
179         read = 0;
180         while (!reader_finish && !kill_test) {
181                 int found;
182
183                 do {
184                         int cpu;
185
186                         found = 0;
187                         for_each_online_cpu(cpu) {
188                                 enum event_status stat;
189
190                                 if (read_events)
191                                         stat = read_event(cpu);
192                                 else
193                                         stat = read_page(cpu);
194
195                                 if (kill_test)
196                                         break;
197                                 if (stat == EVENT_FOUND)
198                                         found = 1;
199                         }
200                 } while (found && !kill_test);
201
202                 set_current_state(TASK_INTERRUPTIBLE);
203                 if (reader_finish)
204                         break;
205
206                 schedule();
207                 __set_current_state(TASK_RUNNING);
208         }
209         reader_finish = 0;
210         complete(&read_done);
211 }
212
213 static void ring_buffer_producer(void)
214 {
215         struct timeval start_tv;
216         struct timeval end_tv;
217         unsigned long long time;
218         unsigned long long entries;
219         unsigned long long overruns;
220         unsigned long missed = 0;
221         unsigned long hit = 0;
222         unsigned long avg;
223         int cnt = 0;
224
225         /*
226          * Hammer the buffer for 10 secs (this may
227          * make the system stall)
228          */
229         trace_printk("Starting ring buffer hammer\n");
230         do_gettimeofday(&start_tv);
231         do {
232                 struct ring_buffer_event *event;
233                 int *entry;
234                 int i;
235
236                 for (i = 0; i < write_iteration; i++) {
237                         event = ring_buffer_lock_reserve(buffer, 10);
238                         if (!event) {
239                                 missed++;
240                         } else {
241                                 hit++;
242                                 entry = ring_buffer_event_data(event);
243                                 *entry = smp_processor_id();
244                                 ring_buffer_unlock_commit(buffer, event);
245                         }
246                 }
247                 do_gettimeofday(&end_tv);
248
249                 cnt++;
250                 if (consumer && !(cnt % wakeup_interval))
251                         wake_up_process(consumer);
252
253 #ifndef CONFIG_PREEMPT
254                 /*
255                  * If we are a non preempt kernel, the 10 second run will
256                  * stop everything while it runs. Instead, we will call
257                  * cond_resched and also add any time that was lost by a
258                  * rescedule.
259                  *
260                  * Do a cond resched at the same frequency we would wake up
261                  * the reader.
262                  */
263                 if (cnt % wakeup_interval)
264                         cond_resched();
265 #endif
266
267         } while (end_tv.tv_sec < (start_tv.tv_sec + RUN_TIME) && !kill_test);
268         trace_printk("End ring buffer hammer\n");
269
270         if (consumer) {
271                 /* Init both completions here to avoid races */
272                 init_completion(&read_start);
273                 init_completion(&read_done);
274                 /* the completions must be visible before the finish var */
275                 smp_wmb();
276                 reader_finish = 1;
277                 /* finish var visible before waking up the consumer */
278                 smp_wmb();
279                 wake_up_process(consumer);
280                 wait_for_completion(&read_done);
281         }
282
283         time = end_tv.tv_sec - start_tv.tv_sec;
284         time *= USEC_PER_SEC;
285         time += (long long)((long)end_tv.tv_usec - (long)start_tv.tv_usec);
286
287         entries = ring_buffer_entries(buffer);
288         overruns = ring_buffer_overruns(buffer);
289
290         if (kill_test)
291                 trace_printk("ERROR!\n");
292
293         if (!disable_reader) {
294                 if (consumer_fifo < 0)
295                         trace_printk("Running Consumer at nice: %d\n",
296                                      consumer_nice);
297                 else
298                         trace_printk("Running Consumer at SCHED_FIFO %d\n",
299                                      consumer_fifo);
300         }
301         if (producer_fifo < 0)
302                 trace_printk("Running Producer at nice: %d\n",
303                              producer_nice);
304         else
305                 trace_printk("Running Producer at SCHED_FIFO %d\n",
306                              producer_fifo);
307
308         /* Let the user know that the test is running at low priority */
309         if (producer_fifo < 0 && consumer_fifo < 0 &&
310             producer_nice == 19 && consumer_nice == 19)
311                 trace_printk("WARNING!!! This test is running at lowest priority.\n");
312
313         trace_printk("Time:     %lld (usecs)\n", time);
314         trace_printk("Overruns: %lld\n", overruns);
315         if (disable_reader)
316                 trace_printk("Read:     (reader disabled)\n");
317         else
318                 trace_printk("Read:     %ld  (by %s)\n", read,
319                         read_events ? "events" : "pages");
320         trace_printk("Entries:  %lld\n", entries);
321         trace_printk("Total:    %lld\n", entries + overruns + read);
322         trace_printk("Missed:   %ld\n", missed);
323         trace_printk("Hit:      %ld\n", hit);
324
325         /* Convert time from usecs to millisecs */
326         do_div(time, USEC_PER_MSEC);
327         if (time)
328                 hit /= (long)time;
329         else
330                 trace_printk("TIME IS ZERO??\n");
331
332         trace_printk("Entries per millisec: %ld\n", hit);
333
334         if (hit) {
335                 /* Calculate the average time in nanosecs */
336                 avg = NSEC_PER_MSEC / hit;
337                 trace_printk("%ld ns per entry\n", avg);
338         }
339
340         if (missed) {
341                 if (time)
342                         missed /= (long)time;
343
344                 trace_printk("Total iterations per millisec: %ld\n",
345                              hit + missed);
346
347                 /* it is possible that hit + missed will overflow and be zero */
348                 if (!(hit + missed)) {
349                         trace_printk("hit + missed overflowed and totalled zero!\n");
350                         hit--; /* make it non zero */
351                 }
352
353                 /* Caculate the average time in nanosecs */
354                 avg = NSEC_PER_MSEC / (hit + missed);
355                 trace_printk("%ld ns per entry\n", avg);
356         }
357 }
358
359 static void wait_to_die(void)
360 {
361         set_current_state(TASK_INTERRUPTIBLE);
362         while (!kthread_should_stop()) {
363                 schedule();
364                 set_current_state(TASK_INTERRUPTIBLE);
365         }
366         __set_current_state(TASK_RUNNING);
367 }
368
369 static int ring_buffer_consumer_thread(void *arg)
370 {
371         while (!kthread_should_stop() && !kill_test) {
372                 complete(&read_start);
373
374                 ring_buffer_consumer();
375
376                 set_current_state(TASK_INTERRUPTIBLE);
377                 if (kthread_should_stop() || kill_test)
378                         break;
379
380                 schedule();
381                 __set_current_state(TASK_RUNNING);
382         }
383         __set_current_state(TASK_RUNNING);
384
385         if (kill_test)
386                 wait_to_die();
387
388         return 0;
389 }
390
391 static int ring_buffer_producer_thread(void *arg)
392 {
393         init_completion(&read_start);
394
395         while (!kthread_should_stop() && !kill_test) {
396                 ring_buffer_reset(buffer);
397
398                 if (consumer) {
399                         smp_wmb();
400                         wake_up_process(consumer);
401                         wait_for_completion(&read_start);
402                 }
403
404                 ring_buffer_producer();
405
406                 trace_printk("Sleeping for 10 secs\n");
407                 set_current_state(TASK_INTERRUPTIBLE);
408                 schedule_timeout(HZ * SLEEP_TIME);
409                 __set_current_state(TASK_RUNNING);
410         }
411
412         if (kill_test)
413                 wait_to_die();
414
415         return 0;
416 }
417
418 static int __init ring_buffer_benchmark_init(void)
419 {
420         int ret;
421
422         /* make a one meg buffer in overwite mode */
423         buffer = ring_buffer_alloc(1000000, RB_FL_OVERWRITE);
424         if (!buffer)
425                 return -ENOMEM;
426
427         if (!disable_reader) {
428                 consumer = kthread_create(ring_buffer_consumer_thread,
429                                           NULL, "rb_consumer");
430                 ret = PTR_ERR(consumer);
431                 if (IS_ERR(consumer))
432                         goto out_fail;
433         }
434
435         producer = kthread_run(ring_buffer_producer_thread,
436                                NULL, "rb_producer");
437         ret = PTR_ERR(producer);
438
439         if (IS_ERR(producer))
440                 goto out_kill;
441
442         /*
443          * Run them as low-prio background tasks by default:
444          */
445         if (!disable_reader) {
446                 if (consumer_fifo >= 0) {
447                         struct sched_param param = {
448                                 .sched_priority = consumer_fifo
449                         };
450                         sched_setscheduler(consumer, SCHED_FIFO, &param);
451                 } else
452                         set_user_nice(consumer, consumer_nice);
453         }
454
455         if (producer_fifo >= 0) {
456                 struct sched_param param = {
457                         .sched_priority = consumer_fifo
458                 };
459                 sched_setscheduler(producer, SCHED_FIFO, &param);
460         } else
461                 set_user_nice(producer, producer_nice);
462
463         return 0;
464
465  out_kill:
466         if (consumer)
467                 kthread_stop(consumer);
468
469  out_fail:
470         ring_buffer_free(buffer);
471         return ret;
472 }
473
474 static void __exit ring_buffer_benchmark_exit(void)
475 {
476         kthread_stop(producer);
477         if (consumer)
478                 kthread_stop(consumer);
479         ring_buffer_free(buffer);
480 }
481
482 module_init(ring_buffer_benchmark_init);
483 module_exit(ring_buffer_benchmark_exit);
484
485 MODULE_AUTHOR("Steven Rostedt");
486 MODULE_DESCRIPTION("ring_buffer_benchmark");
487 MODULE_LICENSE("GPL");