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
  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         if (!bpage)
  88                 return EVENT_DROPPED;
  89
  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
 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
 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
 253                 sched_if_needed(&start_tv, &end_tv);
 254
 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");