2 * Read a file and write the contents to stdout. If a given read takes
3 * longer than 'max_us' time, then we schedule a new thread to handle
4 * the next read. This avoids the coordinated omission problem, where
5 * one request appears to take a long time, but in reality a lot of
6 * requests would have been slow, but we don't notice since new submissions
7 * are not being issued if just 1 is held up.
11 * $ time (./read-to-pipe-async -f randfile.gz | gzip -dc > outfile; sync)
13 * This will read randfile.gz and log the latencies of doing so, while
14 * piping the output to gzip to decompress it. Any latencies over max_us
15 * are logged when they happen, and latency buckets are displayed at the
18 * gcc -Wall -g -O2 -o read-to-pipe-async read-to-pipe-async.c -lpthread
20 * Copyright (C) 2016 Jens Axboe
29 #include <sys/types.h>
39 #include "compiler/compiler.h"
42 static int max_us = 10000;
44 static int separate_writer = 1;
47 #define PLAT_VAL (1 << PLAT_BITS)
48 #define PLAT_GROUP_NR 19
49 #define PLAT_NR (PLAT_GROUP_NR * PLAT_VAL)
50 #define PLAT_LIST_MAX 20
53 #define CHECK_ZERO_OR_ABORT(code) assert(code)
55 #define CHECK_ZERO_OR_ABORT(code) \
57 if (fio_unlikely((code) != 0)) { \
58 log_err("failed checking code %i != 0", (code)); \
65 unsigned int plat[PLAT_NR];
66 unsigned int nr_samples;
72 static double plist[PLAT_LIST_MAX] = { 50.0, 75.0, 90.0, 95.0, 99.0, 99.5, 99.9, 99.99, 99.999, 99.9999, };
79 pthread_mutex_t done_lock;
80 pthread_cond_t done_cond;
84 struct writer_thread {
85 struct flist_head list;
86 struct flist_head done_list;
88 struct thread_data thread;
91 struct reader_thread {
92 struct flist_head list;
93 struct flist_head done_list;
98 struct thread_data thread;
102 struct flist_head list;
108 struct writer_thread *writer;
109 struct reader_thread *reader;
110 pthread_mutex_t lock;
115 static struct reader_thread reader_thread;
116 static struct writer_thread writer_thread;
118 uint64_t utime_since(const struct timespec *s, const struct timespec *e)
123 sec = e->tv_sec - s->tv_sec;
124 usec = (e->tv_nsec - s->tv_nsec) / 1000;
125 if (sec > 0 && usec < 0) {
130 if (sec < 0 || (sec == 0 && usec < 0))
133 ret = sec * 1000000ULL + usec;
138 static struct work_item *find_seq(struct writer_thread *w, int seq)
140 struct work_item *work;
141 struct flist_head *entry;
143 if (flist_empty(&w->list))
146 flist_for_each(entry, &w->list) {
147 work = flist_entry(entry, struct work_item, list);
148 if (work->seq == seq)
155 static unsigned int plat_val_to_idx(unsigned int val)
157 unsigned int msb, error_bits, base, offset;
159 /* Find MSB starting from bit 0 */
163 msb = sizeof(val)*8 - __builtin_clz(val) - 1;
166 * MSB <= (PLAT_BITS-1), cannot be rounded off. Use
167 * all bits of the sample as index
169 if (msb <= PLAT_BITS)
172 /* Compute the number of error bits to discard*/
173 error_bits = msb - PLAT_BITS;
175 /* Compute the number of buckets before the group */
176 base = (error_bits + 1) << PLAT_BITS;
179 * Discard the error bits and apply the mask to find the
180 * index for the buckets in the group
182 offset = (PLAT_VAL - 1) & (val >> error_bits);
184 /* Make sure the index does not exceed (array size - 1) */
185 return (base + offset) < (PLAT_NR - 1) ?
186 (base + offset) : (PLAT_NR - 1);
190 * Convert the given index of the bucket array to the value
191 * represented by the bucket
193 static unsigned int plat_idx_to_val(unsigned int idx)
195 unsigned int error_bits, k, base;
197 assert(idx < PLAT_NR);
199 /* MSB <= (PLAT_BITS-1), cannot be rounded off. Use
200 * all bits of the sample as index */
201 if (idx < (PLAT_VAL << 1))
204 /* Find the group and compute the minimum value of that group */
205 error_bits = (idx >> PLAT_BITS) - 1;
206 base = 1 << (error_bits + PLAT_BITS);
208 /* Find its bucket number of the group */
211 /* Return the mean of the range of the bucket */
212 return base + ((k + 0.5) * (1 << error_bits));
215 static void add_lat(struct stats *s, unsigned int us, const char *name)
225 fprintf(stderr, "%s latency=%u usec\n", name, us);
229 lat_index = plat_val_to_idx(us);
230 __sync_fetch_and_add(&s->plat[lat_index], 1);
231 __sync_fetch_and_add(&s->nr_samples, 1);
234 static int write_work(struct work_item *work)
236 struct timespec s, e;
239 clock_gettime(CLOCK_MONOTONIC, &s);
240 ret = write(STDOUT_FILENO, work->buf, work->buf_size);
243 clock_gettime(CLOCK_MONOTONIC, &e);
244 assert(ret == work->buf_size);
246 add_lat(&work->writer->s, utime_since(&s, &e), "write");
247 return work->seq + 1;
250 static void thread_exiting(struct thread_data *thread)
252 __sync_fetch_and_add(&thread->done, 1);
253 pthread_cond_signal(&thread->done_cond);
256 static void *writer_fn(void *data)
258 struct writer_thread *wt = data;
259 struct work_item *work;
263 while (!(seq < 0) && (!wt->thread.exit || !flist_empty(&wt->list))) {
264 pthread_mutex_lock(&wt->thread.lock);
267 flist_add_tail(&work->list, &wt->done_list);
269 work = find_seq(wt, seq);
271 flist_del_init(&work->list);
273 pthread_cond_wait(&wt->thread.cond, &wt->thread.lock);
275 pthread_mutex_unlock(&wt->thread.lock);
278 seq = write_work(work);
281 thread_exiting(&wt->thread);
285 static void reader_work(struct work_item *work)
287 struct timespec s, e;
293 clock_gettime(CLOCK_MONOTONIC, &s);
295 left = work->buf_size;
299 ret = pread(work->fd, buf, left, off);
301 fprintf(stderr, "zero read\n");
303 } else if (ret < 0) {
304 fprintf(stderr, "errno=%d\n", errno);
312 clock_gettime(CLOCK_MONOTONIC, &e);
314 add_lat(&work->reader->s, utime_since(&s, &e), "read");
316 pthread_cond_signal(&work->cond);
318 if (separate_writer) {
319 pthread_mutex_lock(&work->writer->thread.lock);
320 flist_add_tail(&work->list, &work->writer->list);
321 pthread_mutex_unlock(&work->writer->thread.lock);
322 pthread_cond_signal(&work->writer->thread.cond);
324 struct reader_thread *rt = work->reader;
325 struct work_item *next = NULL;
326 struct flist_head *entry;
329 * Write current work if it matches in sequence.
331 if (work->seq == rt->write_seq)
334 pthread_mutex_lock(&rt->thread.lock);
336 flist_add_tail(&work->list, &rt->done_list);
339 * See if the next work item is here, if so, write it
342 flist_for_each(entry, &rt->done_list) {
343 next = flist_entry(entry, struct work_item, list);
344 if (next->seq == rt->write_seq) {
346 flist_del(&work->list);
351 pthread_mutex_unlock(&rt->thread.lock);
356 __sync_fetch_and_add(&rt->write_seq, 1);
361 static void *reader_one_off(void *data)
367 static void *reader_fn(void *data)
369 struct reader_thread *rt = data;
370 struct work_item *work;
372 while (!rt->thread.exit || !flist_empty(&rt->list)) {
374 pthread_mutex_lock(&rt->thread.lock);
375 if (!flist_empty(&rt->list)) {
376 work = flist_first_entry(&rt->list, struct work_item, list);
377 flist_del_init(&work->list);
379 pthread_cond_wait(&rt->thread.cond, &rt->thread.lock);
380 pthread_mutex_unlock(&rt->thread.lock);
383 __sync_fetch_and_add(&rt->busy, 1);
385 __sync_fetch_and_sub(&rt->busy, 1);
389 thread_exiting(&rt->thread);
393 static void queue_work(struct reader_thread *rt, struct work_item *work)
396 pthread_mutex_lock(&rt->thread.lock);
397 flist_add_tail(&work->list, &rt->list);
398 pthread_mutex_unlock(&rt->thread.lock);
401 pthread_create(&rt->thread.thread, NULL, reader_fn, rt);
402 } else if (!rt->busy && !pthread_mutex_trylock(&rt->thread.lock)) {
403 flist_add_tail(&work->list, &rt->list);
404 pthread_mutex_unlock(&rt->thread.lock);
406 pthread_cond_signal(&rt->thread.cond);
408 int ret = pthread_create(&work->thread, NULL, reader_one_off, work);
410 fprintf(stderr, "pthread_create=%d\n", ret);
412 ret = pthread_detach(work->thread);
414 fprintf(stderr, "pthread_detach=%d\n", ret);
419 static unsigned int calc_percentiles(unsigned int *io_u_plat, unsigned long nr,
420 unsigned int **output)
422 unsigned long sum = 0;
423 unsigned int len, i, j = 0;
424 unsigned int oval_len = 0;
425 unsigned int *ovals = NULL;
429 while (len < PLAT_LIST_MAX && plist[len] != 0.0)
436 * Calculate bucket values, note down max and min values
439 for (i = 0; i < PLAT_NR && !is_last; i++) {
441 while (sum >= (plist[j] / 100.0 * nr)) {
442 assert(plist[j] <= 100.0);
446 ovals = realloc(ovals, oval_len * sizeof(unsigned int));
449 ovals[j] = plat_idx_to_val(i);
450 is_last = (j == len - 1);
462 static void show_latencies(struct stats *s, const char *msg)
464 unsigned int *ovals = NULL;
467 len = calc_percentiles(s->plat, s->nr_samples, &ovals);
469 fprintf(stderr, "Latency percentiles (usec) (%s)\n", msg);
470 for (i = 0; i < len; i++)
471 fprintf(stderr, "\t%2.4fth: %u\n", plist[i], ovals[i]);
477 fprintf(stderr, "\tOver=%u, min=%u, max=%u\n", s->over, s->min, s->max);
480 static void init_thread(struct thread_data *thread)
482 pthread_condattr_t cattr;
485 ret = pthread_condattr_init(&cattr);
486 CHECK_ZERO_OR_ABORT(ret);
487 #ifdef CONFIG_PTHREAD_CONDATTR_SETCLOCK
488 ret = pthread_condattr_setclock(&cattr, CLOCK_MONOTONIC);
489 CHECK_ZERO_OR_ABORT(ret);
491 pthread_cond_init(&thread->cond, &cattr);
492 pthread_cond_init(&thread->done_cond, &cattr);
493 pthread_mutex_init(&thread->lock, NULL);
494 pthread_mutex_init(&thread->done_lock, NULL);
498 static void exit_thread(struct thread_data *thread,
499 void fn(struct writer_thread *),
500 struct writer_thread *wt)
502 __sync_fetch_and_add(&thread->exit, 1);
503 pthread_cond_signal(&thread->cond);
505 while (!thread->done) {
506 pthread_mutex_lock(&thread->done_lock);
511 #ifdef CONFIG_PTHREAD_CONDATTR_SETCLOCK
512 clock_gettime(CLOCK_MONOTONIC, &ts);
514 clock_gettime(CLOCK_REALTIME, &ts);
518 pthread_cond_timedwait(&thread->done_cond, &thread->done_lock, &ts);
521 pthread_cond_wait(&thread->done_cond, &thread->done_lock);
523 pthread_mutex_unlock(&thread->done_lock);
527 static int usage(char *argv[])
529 fprintf(stderr, "%s: [-b blocksize] [-t max usec] [-w separate writer] -f file\n", argv[0]);
533 static int parse_options(int argc, char *argv[])
537 while ((c = getopt(argc, argv, "f:b:t:w:")) != -1) {
542 file = strdup(optarg);
548 max_us = atoi(optarg);
551 separate_writer = atoi(optarg);
552 if (!separate_writer)
553 fprintf(stderr, "inline writing is broken\n");
567 static void prune_done_entries(struct writer_thread *wt)
571 if (flist_empty(&wt->done_list))
574 if (pthread_mutex_trylock(&wt->thread.lock))
577 if (!flist_empty(&wt->done_list))
578 flist_splice_init(&wt->done_list, &list);
579 pthread_mutex_unlock(&wt->thread.lock);
581 while (!flist_empty(&list)) {
582 struct work_item *work;
584 work = flist_first_entry(&list, struct work_item, list);
585 flist_del(&work->list);
587 pthread_cond_destroy(&work->cond);
588 pthread_mutex_destroy(&work->lock);
594 int main(int argc, char *argv[])
596 pthread_condattr_t cattr;
597 struct timespec s, re, we;
598 struct reader_thread *rt;
599 struct writer_thread *wt;
608 if (parse_options(argc, argv))
611 fd = open(file, O_RDONLY);
617 if (fstat(fd, &sb) < 0) {
623 init_thread(&wt->thread);
624 INIT_FLIST_HEAD(&wt->list);
625 INIT_FLIST_HEAD(&wt->done_list);
628 pthread_create(&wt->thread.thread, NULL, writer_fn, wt);
631 init_thread(&rt->thread);
632 INIT_FLIST_HEAD(&rt->list);
633 INIT_FLIST_HEAD(&rt->done_list);
642 ret = pthread_condattr_init(&cattr);
643 CHECK_ZERO_OR_ABORT(ret);
644 #ifdef CONFIG_PTHREAD_CONDATTR_SETCLOCK
645 ret = pthread_condattr_setclock(&cattr, CLOCK_MONOTONIC);
646 CHECK_ZERO_OR_ABORT(ret);
649 clock_gettime(CLOCK_MONOTONIC, &s);
652 struct work_item *work;
656 prune_done_entries(wt);
658 this_len = sb.st_size;
662 work = calloc(1, sizeof(*work));
663 work->buf = malloc(this_len);
664 work->buf_size = this_len;
670 pthread_cond_init(&work->cond, &cattr);
671 pthread_mutex_init(&work->lock, NULL);
673 queue_work(rt, work);
675 #ifdef CONFIG_PTHREAD_CONDATTR_SETCLOCK
676 clock_gettime(CLOCK_MONOTONIC, &ts);
678 clock_gettime(CLOCK_REALTIME, &ts);
680 ts.tv_nsec += max_us * 1000ULL;
681 if (ts.tv_nsec >= 1000000000ULL) {
682 ts.tv_nsec -= 1000000000ULL;
686 pthread_mutex_lock(&work->lock);
687 pthread_cond_timedwait(&work->cond, &work->lock, &ts);
688 pthread_mutex_unlock(&work->lock);
691 sb.st_size -= this_len;
695 exit_thread(&rt->thread, NULL, NULL);
696 clock_gettime(CLOCK_MONOTONIC, &re);
698 exit_thread(&wt->thread, prune_done_entries, wt);
699 clock_gettime(CLOCK_MONOTONIC, &we);
701 show_latencies(&rt->s, "READERS");
702 show_latencies(&wt->s, "WRITERS");
705 elapsed = utime_since(&s, &re);
706 rate = elapsed ? (bytes * 1000UL * 1000UL) / elapsed : 0;
707 fprintf(stderr, "Read rate (KiB/sec) : %lu\n", rate);
708 elapsed = utime_since(&s, &we);
709 rate = elapsed ? (bytes * 1000UL * 1000UL) / elapsed : 0;
710 fprintf(stderr, "Write rate (KiB/sec): %lu\n", rate);