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 = MAX_NICE; 44 static int consumer_nice = MAX_NICE; 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, cpu); 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 } 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 == MAX_NICE && consumer_nice == MAX_NICE) 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 } 382 __set_current_state(TASK_RUNNING); 383 384 if (kill_test) 385 wait_to_die(); 386 387 return 0; 388 } 389 390 static int ring_buffer_producer_thread(void *arg) 391 { 392 init_completion(&read_start); 393 394 while (!kthread_should_stop() && !kill_test) { 395 ring_buffer_reset(buffer); 396 397 if (consumer) { 398 smp_wmb(); 399 wake_up_process(consumer); 400 wait_for_completion(&read_start); 401 } 402 403 ring_buffer_producer(); 404 405 trace_printk("Sleeping for 10 secs\n"); 406 set_current_state(TASK_INTERRUPTIBLE); 407 schedule_timeout(HZ * SLEEP_TIME); 408 } 409 410 if (kill_test) 411 wait_to_die(); 412 413 return 0; 414 } 415 416 static int __init ring_buffer_benchmark_init(void) 417 { 418 int ret; 419 420 /* make a one meg buffer in overwite mode */ 421 buffer = ring_buffer_alloc(1000000, RB_FL_OVERWRITE); 422 if (!buffer) 423 return -ENOMEM; 424 425 if (!disable_reader) { 426 consumer = kthread_create(ring_buffer_consumer_thread, 427 NULL, "rb_consumer"); 428 ret = PTR_ERR(consumer); 429 if (IS_ERR(consumer)) 430 goto out_fail; 431 } 432 433 producer = kthread_run(ring_buffer_producer_thread, 434 NULL, "rb_producer"); 435 ret = PTR_ERR(producer); 436 437 if (IS_ERR(producer)) 438 goto out_kill; 439 440 /* 441 * Run them as low-prio background tasks by default: 442 */ 443 if (!disable_reader) { 444 if (consumer_fifo >= 0) { 445 struct sched_param param = { 446 .sched_priority = consumer_fifo 447 }; 448 sched_setscheduler(consumer, SCHED_FIFO, ¶m); 449 } else 450 set_user_nice(consumer, consumer_nice); 451 } 452 453 if (producer_fifo >= 0) { 454 struct sched_param param = { 455 .sched_priority = consumer_fifo 456 }; 457 sched_setscheduler(producer, SCHED_FIFO, ¶m); 458 } else 459 set_user_nice(producer, producer_nice); 460 461 return 0; 462 463 out_kill: 464 if (consumer) 465 kthread_stop(consumer); 466 467 out_fail: 468 ring_buffer_free(buffer); 469 return ret; 470 } 471 472 static void __exit ring_buffer_benchmark_exit(void) 473 { 474 kthread_stop(producer); 475 if (consumer) 476 kthread_stop(consumer); 477 ring_buffer_free(buffer); 478 } 479 480 module_init(ring_buffer_benchmark_init); 481 module_exit(ring_buffer_benchmark_exit); 482 483 MODULE_AUTHOR("Steven Rostedt"); 484 MODULE_DESCRIPTION("ring_buffer_benchmark"); 485 MODULE_LICENSE("GPL"); 486