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