[fio.git] / t / read-to-pipe-async.c

/*
 * Read a file and write the contents to stdout. If a given read takes
 * longer than 'max_us' time, then we schedule a new thread to handle
 * the next read. This avoids the coordinated omission problem, where
 * one request appears to take a long time, but in reality a lot of
 * requests would have been slow, but we don't notice since new submissions
 * are not being issued if just 1 is held up.
 *
 * One test case:
 *
 * $ time (./read-to-pipe-async -f randfile.gz | gzip -dc > outfile; sync)
 *
 * This will read randfile.gz and log the latencies of doing so, while
 * piping the output to gzip to decompress it. Any latencies over max_us
 * are logged when they happen, and latency buckets are displayed at the
 * end of the run
 *
 * gcc -Wall -g -O2 -o read-to-pipe-async read-to-pipe-async.c -lpthread
 *
 * Copyright (C) 2016 Jens Axboe
 *
 */
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <inttypes.h>
#include <string.h>
#include <pthread.h>
#include <errno.h>
#include <assert.h>
#include <time.h>

#include "../flist.h"

static int bs = 4096;
static int max_us = 10000;
static char *file;
static int separate_writer = 1;

#define PLAT_BITS	8
#define PLAT_VAL	(1 << PLAT_BITS)
#define PLAT_GROUP_NR	19
#define PLAT_NR		(PLAT_GROUP_NR * PLAT_VAL)
#define PLAT_LIST_MAX	20

struct stats {
	unsigned int plat[PLAT_NR];
	unsigned int nr_samples;
	unsigned int max;
	unsigned int min;
	unsigned int over;
};

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, };

struct thread_data {
	int exit;
	int done;
	pthread_mutex_t lock;
	pthread_cond_t cond;
	pthread_mutex_t done_lock;
	pthread_cond_t done_cond;
	pthread_t thread;
};

struct writer_thread {
	struct flist_head list;
	struct flist_head done_list;
	struct stats s;
	struct thread_data thread;
};

struct reader_thread {
	struct flist_head list;
	struct flist_head done_list;
	int started;
	int busy;
	int write_seq;
	struct stats s;
	struct thread_data thread;
};

struct work_item {
	struct flist_head list;
	void *buf;
	size_t buf_size;
	off_t off;
	int fd;
	int seq;
	struct writer_thread *writer;
	struct reader_thread *reader;
	pthread_mutex_t lock;
	pthread_cond_t cond;
	pthread_t thread;
};

static struct reader_thread reader_thread;
static struct writer_thread writer_thread;

uint64_t utime_since(const struct timeval *s, const struct timeval *e)
{
	long sec, usec;
	uint64_t ret;

	sec = e->tv_sec - s->tv_sec;
	usec = e->tv_usec - s->tv_usec;
	if (sec > 0 && usec < 0) {
		sec--;
		usec += 1000000;
	}

	if (sec < 0 || (sec == 0 && usec < 0))
		return 0;

	ret = sec * 1000000ULL + usec;

	return ret;
}

static struct work_item *find_seq(struct writer_thread *w, unsigned int seq)
{
	struct work_item *work;
	struct flist_head *entry;

	if (flist_empty(&w->list))
		return NULL;

	flist_for_each(entry, &w->list) {
		work = flist_entry(entry, struct work_item, list);
		if (work->seq == seq)
			return work;
	}

	return NULL;
}

static unsigned int plat_val_to_idx(unsigned int val)
{
	unsigned int msb, error_bits, base, offset;

	/* Find MSB starting from bit 0 */
	if (val == 0)
		msb = 0;
	else
		msb = sizeof(val)*8 - __builtin_clz(val) - 1;

	/*
	 * MSB <= (PLAT_BITS-1), cannot be rounded off. Use
	 * all bits of the sample as index
	 */
	if (msb <= PLAT_BITS)
		return val;

	/* Compute the number of error bits to discard*/
	error_bits = msb - PLAT_BITS;

	/* Compute the number of buckets before the group */
	base = (error_bits + 1) << PLAT_BITS;

	/*
	 * Discard the error bits and apply the mask to find the
	 * index for the buckets in the group
	 */
	offset = (PLAT_VAL - 1) & (val >> error_bits);

	/* Make sure the index does not exceed (array size - 1) */
	return (base + offset) < (PLAT_NR - 1) ?
		(base + offset) : (PLAT_NR - 1);
}

/*
 * Convert the given index of the bucket array to the value
 * represented by the bucket
 */
static unsigned int plat_idx_to_val(unsigned int idx)
{
	unsigned int error_bits, k, base;

	assert(idx < PLAT_NR);

	/* MSB <= (PLAT_BITS-1), cannot be rounded off. Use
	 * all bits of the sample as index */
	if (idx < (PLAT_VAL << 1))
		return idx;

	/* Find the group and compute the minimum value of that group */
	error_bits = (idx >> PLAT_BITS) - 1;
	base = 1 << (error_bits + PLAT_BITS);

	/* Find its bucket number of the group */
	k = idx % PLAT_VAL;

	/* Return the mean of the range of the bucket */
	return base + ((k + 0.5) * (1 << error_bits));
}

static void add_lat(struct stats *s, unsigned int us, const char *name)
{
	int lat_index = 0;

	if (us > s->max)
		s->max = us;
	if (us < s->min)
		s->min = us;

	if (us > max_us) {
		fprintf(stderr, "%s latency=%u usec\n", name, us);
		s->over++;
	}

	lat_index = plat_val_to_idx(us);
	__sync_fetch_and_add(&s->plat[lat_index], 1);
	__sync_fetch_and_add(&s->nr_samples, 1);
}

static int write_work(struct work_item *work)
{
	struct timeval s, e;
	ssize_t ret;

	gettimeofday(&s, NULL);
	ret = write(STDOUT_FILENO, work->buf, work->buf_size);
	gettimeofday(&e, NULL);
	assert(ret == work->buf_size);

	add_lat(&work->writer->s, utime_since(&s, &e), "write");
	return work->seq + 1;
}

static void *writer_fn(void *data)
{
	struct writer_thread *wt = data;
	struct work_item *work;
	unsigned int seq = 1;

	work = NULL;
	while (!wt->thread.exit || !flist_empty(&wt->list)) {
		pthread_mutex_lock(&wt->thread.lock);

		if (work) {
			flist_add_tail(&work->list, &wt->done_list);
			work = NULL;
		}
	
		work = find_seq(wt, seq);
		if (work)
			flist_del_init(&work->list);
		else
			pthread_cond_wait(&wt->thread.cond, &wt->thread.lock);

		pthread_mutex_unlock(&wt->thread.lock);

		if (work)
			seq = write_work(work);
	}

	wt->thread.done = 1;
	pthread_cond_signal(&wt->thread.done_cond);
	return NULL;
}

static void reader_work(struct work_item *work)
{
	struct timeval s, e;
	ssize_t ret;
	size_t left;
	void *buf;
	off_t off;

	gettimeofday(&s, NULL);

	left = work->buf_size;
	buf = work->buf;
	off = work->off;
	while (left) {
		ret = pread(work->fd, buf, left, off);
		if (!ret) {
			fprintf(stderr, "zero read\n");
			break;
		} else if (ret < 0) {
			fprintf(stderr, "errno=%d\n", errno);
			break;
		}
		left -= ret;
		off += ret;
		buf += ret;
	}

	gettimeofday(&e, NULL);

	add_lat(&work->reader->s, utime_since(&s, &e), "read");

	pthread_cond_signal(&work->cond);

	if (separate_writer) {
		pthread_mutex_lock(&work->writer->thread.lock);
		flist_add_tail(&work->list, &work->writer->list);
		pthread_mutex_unlock(&work->writer->thread.lock);
		pthread_cond_signal(&work->writer->thread.cond);
	} else {
		struct reader_thread *rt = work->reader;
		struct work_item *next = NULL;
		struct flist_head *entry;

		/*
		 * Write current work if it matches in sequence.
		 */
		if (work->seq == rt->write_seq)
			goto write_it;

		pthread_mutex_lock(&rt->thread.lock);

		flist_add_tail(&work->list, &rt->done_list);

		/*
		 * See if the next work item is here, if so, write it
		 */
		work = NULL;
		flist_for_each(entry, &rt->done_list) {
			next = flist_entry(entry, struct work_item, list);
			if (next->seq == rt->write_seq) {
				work = next;
				flist_del(&work->list);
				break;
			}
		}

		pthread_mutex_unlock(&rt->thread.lock);
	
		if (work) {
write_it:
			write_work(work);
			__sync_fetch_and_add(&rt->write_seq, 1);
		}
	}
}

static void *reader_one_off(void *data)
{
	reader_work(data);
	return NULL;
}

static void *reader_fn(void *data)
{
	struct reader_thread *rt = data;
	struct work_item *work;

	while (!rt->thread.exit || !flist_empty(&rt->list)) {
		work = NULL;
		pthread_mutex_lock(&rt->thread.lock);
		if (!flist_empty(&rt->list)) {
			work = flist_first_entry(&rt->list, struct work_item, list);
			flist_del_init(&work->list);
		} else
			pthread_cond_wait(&rt->thread.cond, &rt->thread.lock);
		pthread_mutex_unlock(&rt->thread.lock);

		if (work) {
			rt->busy = 1;
			reader_work(work);
			rt->busy = 0;
		}
	}

	rt->thread.done = 1;
	pthread_cond_signal(&rt->thread.done_cond);
	return NULL;
}

static void queue_work(struct reader_thread *rt, struct work_item *work)
{
	if (!rt->started) {
		pthread_mutex_lock(&rt->thread.lock);
		flist_add_tail(&work->list, &rt->list);
		pthread_mutex_unlock(&rt->thread.lock);

		rt->started = 1;
		pthread_create(&rt->thread.thread, NULL, reader_fn, rt);
	} else if (!rt->busy && !pthread_mutex_trylock(&rt->thread.lock)) {
		flist_add_tail(&work->list, &rt->list);
		pthread_mutex_unlock(&rt->thread.lock);

		pthread_cond_signal(&rt->thread.cond);
	} else {
		int ret = pthread_create(&work->thread, NULL, reader_one_off, work);
		if (ret)
			fprintf(stderr, "pthread_create=%d\n", ret);
		else
			pthread_detach(work->thread);
	}
}

static unsigned int calc_percentiles(unsigned int *io_u_plat, unsigned long nr,
				     unsigned int **output)
{
	unsigned long sum = 0;
	unsigned int len, i, j = 0;
	unsigned int oval_len = 0;
	unsigned int *ovals = NULL;
	int is_last;

	len = 0;
	while (len < PLAT_LIST_MAX && plist[len] != 0.0)
		len++;

	if (!len)
		return 0;

	/*
	 * Calculate bucket values, note down max and min values
	 */
	is_last = 0;
	for (i = 0; i < PLAT_NR && !is_last; i++) {
		sum += io_u_plat[i];
		while (sum >= (plist[j] / 100.0 * nr)) {
			assert(plist[j] <= 100.0);

			if (j == oval_len) {
				oval_len += 100;
				ovals = realloc(ovals, oval_len * sizeof(unsigned int));
			}

			ovals[j] = plat_idx_to_val(i);
			is_last = (j == len - 1);
			if (is_last)
				break;

			j++;
		}
	}

	*output = ovals;
	return len;
}

static void show_latencies(struct stats *s, const char *msg)
{
	unsigned int *ovals = NULL;
	unsigned int len, i;

	len = calc_percentiles(s->plat, s->nr_samples, &ovals);
	if (len) {
		fprintf(stderr, "Latency percentiles (usec) (%s)\n", msg);
		for (i = 0; i < len; i++)
			fprintf(stderr, "\t%2.4fth: %u\n", plist[i], ovals[i]);
	}

	if (ovals)
		free(ovals);

	fprintf(stderr, "\tOver=%u, min=%u, max=%u\n", s->over, s->min, s->max);
}

static void init_thread(struct thread_data *thread)
{
	pthread_cond_init(&thread->cond, NULL);
	pthread_cond_init(&thread->done_cond, NULL);
	pthread_mutex_init(&thread->lock, NULL);
	pthread_mutex_init(&thread->done_lock, NULL);
	thread->exit = 0;
}

static void exit_thread(struct thread_data *thread,
			void fn(struct writer_thread *),
			struct writer_thread *wt)
{
	thread->exit = 1;
	pthread_cond_signal(&thread->cond);

	while (!thread->done) {
		pthread_mutex_lock(&thread->done_lock);

		if (fn) {
			struct timespec t;

			clock_gettime(CLOCK_REALTIME, &t);
			t.tv_sec++;


			pthread_cond_timedwait(&thread->done_cond, &thread->done_lock, &t);
			fn(wt);
		} else
			pthread_cond_wait(&thread->done_cond, &thread->done_lock);

		pthread_mutex_unlock(&thread->done_lock);
	}
}

static int usage(char *argv[])
{
	fprintf(stderr, "%s: [-b blocksize] [-t max usec] [-w separate writer] -f file\n", argv[0]);
	return 1;
}

static int parse_options(int argc, char *argv[])
{
	int c;

	while ((c = getopt(argc, argv, "f:b:t:w:")) != -1) {
		switch (c) {
		case 'f':
			file = strdup(optarg);
			break;
		case 'b':
			bs = atoi(optarg);
			break;
		case 't':
			max_us = atoi(optarg);
			break;
		case 'w':
			separate_writer = atoi(optarg);
			if (!separate_writer)
				fprintf(stderr, "inline writing is broken\n");
			break;
		case '?':
		default:
			return usage(argv);
		}
	}

	if (!file)
		return usage(argv);

	return 0;
}

static void prune_done_entries(struct writer_thread *wt)
{
	FLIST_HEAD(list);

	if (flist_empty(&wt->done_list))
		return;

	if (pthread_mutex_trylock(&wt->thread.lock))
		return;

	if (!flist_empty(&wt->done_list))
		flist_splice_init(&wt->done_list, &list);
	pthread_mutex_unlock(&wt->thread.lock);

	while (!flist_empty(&list)) {
		struct work_item *work;

		work = flist_first_entry(&list, struct work_item, list);
		flist_del(&work->list);

		pthread_cond_destroy(&work->cond);
		pthread_mutex_destroy(&work->lock);
		free(work->buf);
		free(work);
	}
}

int main(int argc, char *argv[])
{
	struct timeval s, re, we;
	struct reader_thread *rt;
	struct writer_thread *wt;
	unsigned long rate;
	struct stat sb;
	size_t bytes;
	off_t off;
	int fd, seq;

	if (parse_options(argc, argv))
		return 1;

	fd = open(file, O_RDONLY);
	if (fd < 0) {
		perror("open");
		return 2;
	}

	if (fstat(fd, &sb) < 0) {
		perror("stat");
		return 3;
	}

	wt = &writer_thread;
	init_thread(&wt->thread);
	INIT_FLIST_HEAD(&wt->list);
	INIT_FLIST_HEAD(&wt->done_list);
	wt->s.max = 0;
	wt->s.min = -1U;
	pthread_create(&wt->thread.thread, NULL, writer_fn, wt);

	rt = &reader_thread;
	init_thread(&rt->thread);
	INIT_FLIST_HEAD(&rt->list);
	INIT_FLIST_HEAD(&rt->done_list);
	rt->s.max = 0;
	rt->s.min = -1U;
	rt->write_seq = 1;

	off = 0;
	seq = 0;
	bytes = 0;

	gettimeofday(&s, NULL);

	while (sb.st_size) {
		struct work_item *work;
		size_t this_len;
		struct timespec t;

		prune_done_entries(wt);

		this_len = sb.st_size;
		if (this_len > bs)
			this_len = bs;

		work = calloc(1, sizeof(*work));
		work->buf = malloc(this_len);
		work->buf_size = this_len;
		work->off = off;
		work->fd = fd;
		work->seq = ++seq;
		work->writer = wt;
		work->reader = rt;
		pthread_cond_init(&work->cond, NULL);
		pthread_mutex_init(&work->lock, NULL);

		queue_work(rt, work);

		clock_gettime(CLOCK_REALTIME, &t);
		t.tv_nsec += max_us * 1000ULL;
		if (t.tv_nsec >= 1000000000ULL) {
			t.tv_nsec -= 1000000000ULL;
			t.tv_sec++;
		}

		pthread_mutex_lock(&work->lock);
		pthread_cond_timedwait(&work->cond, &work->lock, &t);
		pthread_mutex_unlock(&work->lock);

		off += this_len;
		sb.st_size -= this_len;
		bytes += this_len;
	}

	exit_thread(&rt->thread, NULL, NULL);
	gettimeofday(&re, NULL);

	exit_thread(&wt->thread, prune_done_entries, wt);
	gettimeofday(&we, NULL);

	show_latencies(&rt->s, "READERS");
	show_latencies(&wt->s, "WRITERS");

	bytes /= 1024;
	rate = (bytes * 1000UL * 1000UL) / utime_since(&s, &re);
	fprintf(stderr, "Read rate (KB/sec) : %lu\n", rate);
	rate = (bytes * 1000UL * 1000UL) / utime_since(&s, &we);
	fprintf(stderr, "Write rate (KB/sec): %lu\n", rate);

	close(fd);
	return 0;
}
Commit	Line	Data
6d5a9bc2 JA	1	/*
	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.
	8	*
	9	* One test case:
	10	*
	11	* $ time (./read-to-pipe-async -f randfile.gz \| gzip -dc > outfile; sync)
	12	*
	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
	16	* end of the run
	17	*
	18	* gcc -Wall -g -O2 -o read-to-pipe-async read-to-pipe-async.c -lpthread
	19	*
	20	* Copyright (C) 2016 Jens Axboe
	21	*
	22	*/
	23	#include <stdio.h>
	24	#include <stdlib.h>
	25	#include <unistd.h>
	26	#include <fcntl.h>
	27	#include <sys/time.h>
	28	#include <sys/types.h>
	29	#include <sys/stat.h>
	30	#include <inttypes.h>
	31	#include <string.h>
	32	#include <pthread.h>
	33	#include <errno.h>
	34	#include <assert.h>
27386101	35	#include <time.h>
6d5a9bc2 JA	36
	37	#include "../flist.h"
	38
	39	static int bs = 4096;
	40	static int max_us = 10000;
	41	static char *file;
	42	static int separate_writer = 1;
	43
	44	#define PLAT_BITS 8
	45	#define PLAT_VAL (1 << PLAT_BITS)
	46	#define PLAT_GROUP_NR 19
	47	#define PLAT_NR (PLAT_GROUP_NR * PLAT_VAL)
	48	#define PLAT_LIST_MAX 20
	49
	50	struct stats {
	51	unsigned int plat[PLAT_NR];
	52	unsigned int nr_samples;
	53	unsigned int max;
	54	unsigned int min;
	55	unsigned int over;
	56	};
	57
	58	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, };
	59
	60	struct thread_data {
	61	int exit;
	62	int done;
	63	pthread_mutex_t lock;
	64	pthread_cond_t cond;
	65	pthread_mutex_t done_lock;
	66	pthread_cond_t done_cond;
	67	pthread_t thread;
	68	};
	69
	70	struct writer_thread {
	71	struct flist_head list;
	72	struct flist_head done_list;
	73	struct stats s;
	74	struct thread_data thread;
	75	};
	76
	77	struct reader_thread {
	78	struct flist_head list;
	79	struct flist_head done_list;
	80	int started;
	81	int busy;
	82	int write_seq;
	83	struct stats s;
	84	struct thread_data thread;
	85	};
	86
	87	struct work_item {
	88	struct flist_head list;
	89	void *buf;
	90	size_t buf_size;
	91	off_t off;
	92	int fd;
	93	int seq;
	94	struct writer_thread *writer;
	95	struct reader_thread *reader;
	96	pthread_mutex_t lock;
	97	pthread_cond_t cond;
	98	pthread_t thread;
	99	};
100
101	static struct reader_thread reader_thread;
102	static struct writer_thread writer_thread;
103
104	uint64_t utime_since(const struct timeval s, const struct timeval e)
105	{
106	long sec, usec;
107	uint64_t ret;
108
109	sec = e->tv_sec - s->tv_sec;
110	usec = e->tv_usec - s->tv_usec;
111	if (sec > 0 && usec < 0) {
112	sec--;
113	usec += 1000000;
114	}
115
116	if (sec < 0 \|\| (sec == 0 && usec < 0))
117	return 0;
118
119	ret = sec * 1000000ULL + usec;
120
121	return ret;
122	}
123
124	static struct work_item find_seq(struct writer_thread w, unsigned int seq)
125	{
126	struct work_item *work;
127	struct flist_head *entry;
128
129	if (flist_empty(&w->list))
130	return NULL;
131
132	flist_for_each(entry, &w->list) {
133	work = flist_entry(entry, struct work_item, list);
134	if (work->seq == seq)
135	return work;
136	}
137
138	return NULL;
139	}
140
141	static unsigned int plat_val_to_idx(unsigned int val)
142	{
143	unsigned int msb, error_bits, base, offset;
144
145	/* Find MSB starting from bit 0 */
146	if (val == 0)
147	msb = 0;
148	else
149	msb = sizeof(val)*8 - __builtin_clz(val) - 1;
150
151	/*
152	* MSB <= (PLAT_BITS-1), cannot be rounded off. Use
153	* all bits of the sample as index
154	*/
155	if (msb <= PLAT_BITS)
156	return val;
157
158	/* Compute the number of error bits to discard*/
159	error_bits = msb - PLAT_BITS;
160
161	/* Compute the number of buckets before the group */
162	base = (error_bits + 1) << PLAT_BITS;
163
164	/*
165	* Discard the error bits and apply the mask to find the
166	* index for the buckets in the group
167	*/
168	offset = (PLAT_VAL - 1) & (val >> error_bits);
169
170	/* Make sure the index does not exceed (array size - 1) */
171	return (base + offset) < (PLAT_NR - 1) ?
172	(base + offset) : (PLAT_NR - 1);
173	}
174
175	/*
176	* Convert the given index of the bucket array to the value
177	* represented by the bucket
178	*/
179	static unsigned int plat_idx_to_val(unsigned int idx)
180	{
181	unsigned int error_bits, k, base;
182
183	assert(idx < PLAT_NR);
184
185	/* MSB <= (PLAT_BITS-1), cannot be rounded off. Use
186	* all bits of the sample as index */
187	if (idx < (PLAT_VAL << 1))
188	return idx;
189
190	/* Find the group and compute the minimum value of that group */
191	error_bits = (idx >> PLAT_BITS) - 1;
192	base = 1 << (error_bits + PLAT_BITS);
193
194	/* Find its bucket number of the group */
195	k = idx % PLAT_VAL;
196
197	/* Return the mean of the range of the bucket */
198	return base + ((k + 0.5) * (1 << error_bits));
199	}
200
201	static void add_lat(struct stats s, unsigned int us, const char name)
202	{
203	int lat_index = 0;
204
205	if (us > s->max)
206	s->max = us;
207	if (us < s->min)
208	s->min = us;
209
210	if (us > max_us) {
211	fprintf(stderr, "%s latency=%u usec\n", name, us);
212	s->over++;
213	}
214
215	lat_index = plat_val_to_idx(us);
216	__sync_fetch_and_add(&s->plat[lat_index], 1);
217	__sync_fetch_and_add(&s->nr_samples, 1);
218	}
219
220	static int write_work(struct work_item *work)
221	{
222	struct timeval s, e;
223	ssize_t ret;
224
225	gettimeofday(&s, NULL);
226	ret = write(STDOUT_FILENO, work->buf, work->buf_size);
227	gettimeofday(&e, NULL);
228	assert(ret == work->buf_size);
229
230	add_lat(&work->writer->s, utime_since(&s, &e), "write");
231	return work->seq + 1;
232	}
233
234	static void writer_fn(void data)
235	{
236	struct writer_thread *wt = data;
237	struct work_item *work;
238	unsigned int seq = 1;
239
240	work = NULL;
241	while (!wt->thread.exit \|\| !flist_empty(&wt->list)) {
242	pthread_mutex_lock(&wt->thread.lock);
243
244	if (work) {
245	flist_add_tail(&work->list, &wt->done_list);
246	work = NULL;
247	}
248
249	work = find_seq(wt, seq);
250	if (work)
251	flist_del_init(&work->list);
252	else
253	pthread_cond_wait(&wt->thread.cond, &wt->thread.lock);
254
255	pthread_mutex_unlock(&wt->thread.lock);
256
257	if (work)
258	seq = write_work(work);
259	}
260
261	wt->thread.done = 1;
262	pthread_cond_signal(&wt->thread.done_cond);
263	return NULL;
264	}
265
266	static void reader_work(struct work_item *work)
267	{
268	struct timeval s, e;
269	ssize_t ret;
270	size_t left;
271	void *buf;
272	off_t off;
273
274	gettimeofday(&s, NULL);
275
276	left = work->buf_size;
277	buf = work->buf;
278	off = work->off;
279	while (left) {
280	ret = pread(work->fd, buf, left, off);
281	if (!ret) {
282	fprintf(stderr, "zero read\n");
283	break;
284	} else if (ret < 0) {
285	fprintf(stderr, "errno=%d\n", errno);
286	break;
287	}
288	left -= ret;
289	off += ret;
290	buf += ret;
291	}
292
293	gettimeofday(&e, NULL);
294
295	add_lat(&work->reader->s, utime_since(&s, &e), "read");
296
297	pthread_cond_signal(&work->cond);
298
299	if (separate_writer) {
300	pthread_mutex_lock(&work->writer->thread.lock);
301	flist_add_tail(&work->list, &work->writer->list);
302	pthread_mutex_unlock(&work->writer->thread.lock);
303	pthread_cond_signal(&work->writer->thread.cond);
304	} else {
305	struct reader_thread *rt = work->reader;
306	struct work_item *next = NULL;
307	struct flist_head *entry;
308
309	/*
310	* Write current work if it matches in sequence.
311	*/
312	if (work->seq == rt->write_seq)
313	goto write_it;
314
315	pthread_mutex_lock(&rt->thread.lock);
316
317	flist_add_tail(&work->list, &rt->done_list);
318
319	/*
320	* See if the next work item is here, if so, write it
321	*/
322	work = NULL;
323	flist_for_each(entry, &rt->done_list) {
324	next = flist_entry(entry, struct work_item, list);
325	if (next->seq == rt->write_seq) {
326	work = next;
327	flist_del(&work->list);
328	break;
329	}
330	}
331
332	pthread_mutex_unlock(&rt->thread.lock);
333
334	if (work) {
335	write_it:
336	write_work(work);
337	__sync_fetch_and_add(&rt->write_seq, 1);
338	}
339	}
340	}
341
342	static void reader_one_off(void data)
343	{
344	reader_work(data);
345	return NULL;
346	}
347
348	static void reader_fn(void data)
349	{
350	struct reader_thread *rt = data;
351	struct work_item *work;
352
353	while (!rt->thread.exit \|\| !flist_empty(&rt->list)) {
354	work = NULL;
355	pthread_mutex_lock(&rt->thread.lock);
356	if (!flist_empty(&rt->list)) {
357	work = flist_first_entry(&rt->list, struct work_item, list);
358	flist_del_init(&work->list);
359	} else
360	pthread_cond_wait(&rt->thread.cond, &rt->thread.lock);
361	pthread_mutex_unlock(&rt->thread.lock);
362
363	if (work) {
364	rt->busy = 1;
365	reader_work(work);
366	rt->busy = 0;
367	}
368	}
369
370	rt->thread.done = 1;
371	pthread_cond_signal(&rt->thread.done_cond);
372	return NULL;
373	}
374
375	static void queue_work(struct reader_thread rt, struct work_item work)
376	{
377	if (!rt->started) {
378	pthread_mutex_lock(&rt->thread.lock);
379	flist_add_tail(&work->list, &rt->list);
380	pthread_mutex_unlock(&rt->thread.lock);
381
382	rt->started = 1;
383	pthread_create(&rt->thread.thread, NULL, reader_fn, rt);
384	} else if (!rt->busy && !pthread_mutex_trylock(&rt->thread.lock)) {
385	flist_add_tail(&work->list, &rt->list);
386	pthread_mutex_unlock(&rt->thread.lock);
387
388	pthread_cond_signal(&rt->thread.cond);
389	} else {
390	int ret = pthread_create(&work->thread, NULL, reader_one_off, work);
391	if (ret)
392	fprintf(stderr, "pthread_create=%d\n", ret);
393	else
394	pthread_detach(work->thread);
395	}
396	}
397
398	static unsigned int calc_percentiles(unsigned int *io_u_plat, unsigned long nr,
399	unsigned int **output)
400	{
401	unsigned long sum = 0;
402	unsigned int len, i, j = 0;
403	unsigned int oval_len = 0;
404	unsigned int *ovals = NULL;
405	int is_last;
406
407	len = 0;
408	while (len < PLAT_LIST_MAX && plist[len] != 0.0)
409	len++;
410
411	if (!len)
412	return 0;
413
414	/*
415	* Calculate bucket values, note down max and min values
416	*/
417	is_last = 0;
418	for (i = 0; i < PLAT_NR && !is_last; i++) {
419	sum += io_u_plat[i];
420	while (sum >= (plist[j] / 100.0 * nr)) {
421	assert(plist[j] <= 100.0);
422
423	if (j == oval_len) {
424	oval_len += 100;
425	ovals = realloc(ovals, oval_len * sizeof(unsigned int));
426	}
427
428	ovals[j] = plat_idx_to_val(i);
429	is_last = (j == len - 1);
430	if (is_last)
431	break;
432
433	j++;
434	}
435	}
436
437	*output = ovals;
438	return len;
439	}
440
441	static void show_latencies(struct stats s, const char msg)
442	{
443	unsigned int *ovals = NULL;
444	unsigned int len, i;
445
446	len = calc_percentiles(s->plat, s->nr_samples, &ovals);
447	if (len) {
448	fprintf(stderr, "Latency percentiles (usec) (%s)\n", msg);
449	for (i = 0; i < len; i++)
450	fprintf(stderr, "\t%2.4fth: %u\n", plist[i], ovals[i]);
451	}
452
453	if (ovals)
454	free(ovals);
455
456	fprintf(stderr, "\tOver=%u, min=%u, max=%u\n", s->over, s->min, s->max);
457	}
458
459	static void init_thread(struct thread_data *thread)
460	{
461	pthread_cond_init(&thread->cond, NULL);
462	pthread_cond_init(&thread->done_cond, NULL);
463	pthread_mutex_init(&thread->lock, NULL);
464	pthread_mutex_init(&thread->done_lock, NULL);
465	thread->exit = 0;
466	}
467
468	static void exit_thread(struct thread_data *thread,
469	void fn(struct writer_thread *),
470	struct writer_thread *wt)
471	{
472	thread->exit = 1;
473	pthread_cond_signal(&thread->cond);
474
475	while (!thread->done) {
476	pthread_mutex_lock(&thread->done_lock);
477
478	if (fn) {
479	struct timespec t;
480
481	clock_gettime(CLOCK_REALTIME, &t);
482	t.tv_sec++;
483
484
485	pthread_cond_timedwait(&thread->done_cond, &thread->done_lock, &t);
486	fn(wt);
487	} else
488	pthread_cond_wait(&thread->done_cond, &thread->done_lock);
489
490	pthread_mutex_unlock(&thread->done_lock);
491	}
492	}
493
494	static int usage(char *argv[])
495	{
496	fprintf(stderr, "%s: [-b blocksize] [-t max usec] [-w separate writer] -f file\n", argv[0]);
497	return 1;
498	}
499
500	static int parse_options(int argc, char *argv[])
501	{
502	int c;
503
504	while ((c = getopt(argc, argv, "f:b:t:w:")) != -1) {
505	switch (c) {
506	case 'f':
507	file = strdup(optarg);
508	break;
509	case 'b':
510	bs = atoi(optarg);
511	break;
512	case 't':
513	max_us = atoi(optarg);
514	break;
515	case 'w':
516	separate_writer = atoi(optarg);
517	if (!separate_writer)
518	fprintf(stderr, "inline writing is broken\n");
519	break;
520	case '?':
521	default:
522	return usage(argv);
523	}
524	}
525
526	if (!file)
527	return usage(argv);
528
529	return 0;
530	}
531
532	static void prune_done_entries(struct writer_thread *wt)
533	{
534	FLIST_HEAD(list);
535
536	if (flist_empty(&wt->done_list))
537	return;
538
539	if (pthread_mutex_trylock(&wt->thread.lock))
540	return;
541
542	if (!flist_empty(&wt->done_list))
543	flist_splice_init(&wt->done_list, &list);
544	pthread_mutex_unlock(&wt->thread.lock);
545
546	while (!flist_empty(&list)) {
547	struct work_item *work;
548
549	work = flist_first_entry(&list, struct work_item, list);
550	flist_del(&work->list);
551
552	pthread_cond_destroy(&work->cond);
553	pthread_mutex_destroy(&work->lock);
554	free(work->buf);
555	free(work);
556	}
557	}
558
559	int main(int argc, char *argv[])
560	{
561	struct timeval s, re, we;
562	struct reader_thread *rt;
563	struct writer_thread *wt;
564	unsigned long rate;
565	struct stat sb;
566	size_t bytes;
567	off_t off;
568	int fd, seq;
569
570	if (parse_options(argc, argv))
571	return 1;
572
573	fd = open(file, O_RDONLY);
574	if (fd < 0) {
575	perror("open");
576	return 2;
577	}
578
579	if (fstat(fd, &sb) < 0) {
580	perror("stat");
581	return 3;
582	}
583
584	wt = &writer_thread;
585	init_thread(&wt->thread);
586	INIT_FLIST_HEAD(&wt->list);
587	INIT_FLIST_HEAD(&wt->done_list);
588	wt->s.max = 0;
589	wt->s.min = -1U;
590	pthread_create(&wt->thread.thread, NULL, writer_fn, wt);
591
592	rt = &reader_thread;
593	init_thread(&rt->thread);
594	INIT_FLIST_HEAD(&rt->list);
595	INIT_FLIST_HEAD(&rt->done_list);
596	rt->s.max = 0;
597	rt->s.min = -1U;
598	rt->write_seq = 1;
599
600	off = 0;
601	seq = 0;
602	bytes = 0;
603
604	gettimeofday(&s, NULL);
605
606	while (sb.st_size) {
607	struct work_item *work;
608	size_t this_len;
609	struct timespec t;
610
611	prune_done_entries(wt);
612
613	this_len = sb.st_size;
614	if (this_len > bs)
615	this_len = bs;
616
617	work = calloc(1, sizeof(*work));
618	work->buf = malloc(this_len);
619	work->buf_size = this_len;
620	work->off = off;
621	work->fd = fd;
622	work->seq = ++seq;
623	work->writer = wt;
624	work->reader = rt;
625	pthread_cond_init(&work->cond, NULL);
626	pthread_mutex_init(&work->lock, NULL);
627
628	queue_work(rt, work);
629
630	clock_gettime(CLOCK_REALTIME, &t);
631	t.tv_nsec += max_us * 1000ULL;
632	if (t.tv_nsec >= 1000000000ULL) {
633	t.tv_nsec -= 1000000000ULL;
634	t.tv_sec++;
635	}
636
637	pthread_mutex_lock(&work->lock);
638	pthread_cond_timedwait(&work->cond, &work->lock, &t);
639	pthread_mutex_unlock(&work->lock);
640
641	off += this_len;
642	sb.st_size -= this_len;
643	bytes += this_len;
644	}
645
646	exit_thread(&rt->thread, NULL, NULL);
647	gettimeofday(&re, NULL);
648
649	exit_thread(&wt->thread, prune_done_entries, wt);
650	gettimeofday(&we, NULL);
651
652	show_latencies(&rt->s, "READERS");
653	show_latencies(&wt->s, "WRITERS");
654
655	bytes /= 1024;
656	rate = (bytes * 1000UL * 1000UL) / utime_since(&s, &re);
657	fprintf(stderr, "Read rate (KB/sec) : %lu\n", rate);
658	rate = (bytes * 1000UL * 1000UL) / utime_since(&s, &we);
659	fprintf(stderr, "Write rate (KB/sec): %lu\n", rate);
660
661	close(fd);
662	return 0;
663	}