[fio.git] / t / aio-ring.c

/*
 * gcc -D_GNU_SOURCE -Wall -O2 -o aio-ring aio-ring.c  -lpthread -laio
 */
#include <stdio.h>
#include <errno.h>
#include <assert.h>
#include <stdlib.h>
#include <stddef.h>
#include <signal.h>
#include <inttypes.h>

#include <sys/types.h>
#include <sys/stat.h>
#include <sys/ioctl.h>
#include <sys/syscall.h>
#include <sys/resource.h>
#include <linux/fs.h>
#include <fcntl.h>
#include <unistd.h>
#include <libaio.h>
#include <string.h>
#include <pthread.h>
#include <sched.h>

#define IOCB_FLAG_HIPRI		(1 << 2)

#define IOCTX_FLAG_IOPOLL	(1 << 1)
#define IOCTX_FLAG_USERIOCB	(1 << 0)
#define IOCTX_FLAG_FIXEDBUFS	(1 << 2)
#define IOCTX_FLAG_SCQRING	(1 << 3)	/* Use SQ/CQ rings */
#define IOCTX_FLAG_SQTHREAD	(1 << 4)	/* Use SQ thread */
#define IOCTX_FLAG_SQWQ		(1 << 5)	/* Use SQ wq */

#define barrier()	__asm__ __volatile__("": : :"memory")

#define min(a, b)		((a < b) ? (a) : (b))

typedef uint32_t u32;

struct aio_iocb_ring {
	union {
		struct {
			u32 head, tail;
			u32 nr_events;
			u32 sq_thread_cpu;
		};
		struct iocb pad_iocb;
	};
	struct iocb iocbs[0];
};

struct aio_io_event_ring {
	union {
		struct {
			u32 head, tail;
			u32 nr_events;
		};
		struct io_event pad_event;
	};
	struct io_event events[0];
};

#define IORING_FLAG_SUBMIT	(1 << 0)
#define IORING_FLAG_GETEVENTS	(1 << 1)

#define container_of(ptr, type, member)  ({				\
	const __typeof__( ((type *)0)->member ) *__mptr = (ptr);	\
	(type *)( (char *)__mptr - offsetof(type,member) );})

#define DEPTH			32
#define RING_SIZE		(DEPTH + 1)

#define BATCH_SUBMIT		8
#define BATCH_COMPLETE		8

#define BS			4096

struct submitter {
	pthread_t thread;
	unsigned long max_blocks;
	io_context_t ioc;
	struct drand48_data rand;
	struct aio_iocb_ring *sq_ring;
	struct aio_io_event_ring *cq_ring;
	int inflight;
	unsigned long reaps;
	unsigned long done;
	unsigned long calls;
	volatile int finish;
	char filename[128];
};

static struct submitter submitters[1];
static volatile int finish;

static int polled = 1;		/* use IO polling */
static int fixedbufs = 1;	/* use fixed user buffers */
static int buffered = 0;	/* use buffered IO, not O_DIRECT */
static int sq_thread = 0;
static int sq_thread_cpu = 0;

static int io_setup2(unsigned int nr_events, unsigned int flags,
		     struct iocb *iocbs, struct aio_iocb_ring *sq_ring,
		     struct aio_io_event_ring *cq_ring, io_context_t *ctx_idp)
{
	return syscall(335, nr_events, flags, iocbs, sq_ring, cq_ring, ctx_idp);
}

static int io_ring_enter(io_context_t ctx, unsigned int to_submit,
			 unsigned int min_complete, unsigned int flags)
{
	return syscall(336, ctx, to_submit, min_complete, flags);
}

static int gettid(void)
{
	return syscall(__NR_gettid);
}

static void init_io(struct submitter *s, int fd, struct iocb *iocb)
{
	unsigned long offset;
	long r;

	lrand48_r(&s->rand, &r);
	offset = (r % (s->max_blocks - 1)) * BS;

	iocb->aio_fildes = fd;
	iocb->aio_lio_opcode = IO_CMD_PREAD;
	iocb->u.c.offset = offset;
	if (polled)
		iocb->u.c.flags = IOCB_FLAG_HIPRI;
	if (!fixedbufs)
		iocb->u.c.nbytes = BS;
}

static int prep_more_ios(struct submitter *s, int fd, int max_ios)
{
	struct aio_iocb_ring *ring = s->sq_ring;
	struct iocb *iocb;
	u32 tail, next_tail, prepped = 0;

	next_tail = tail = ring->tail;
	do {
		next_tail++;
		if (next_tail == ring->nr_events)
			next_tail = 0;

		barrier();
		if (next_tail == ring->head)
			break;

		iocb = &s->sq_ring->iocbs[tail];
		init_io(s, fd, iocb);
		prepped++;
		tail = next_tail;
	} while (prepped < max_ios);

	if (ring->tail != tail) {
		/* order tail store with writes to iocbs above */
		barrier();
		ring->tail = tail;
		barrier();
	}
	return prepped;
}

static int get_file_size(int fd, unsigned long *blocks)
{
	struct stat st;

	if (fstat(fd, &st) < 0)
		return -1;
	if (S_ISBLK(st.st_mode)) {
		unsigned long long bytes;

		if (ioctl(fd, BLKGETSIZE64, &bytes) != 0)
			return -1;

		*blocks = bytes / BS;
		return 0;
	} else if (S_ISREG(st.st_mode)) {
		*blocks = st.st_size / BS;
		return 0;
	}

	return -1;
}

static int reap_events(struct submitter *s)
{
	struct aio_io_event_ring *ring = s->cq_ring;
	struct io_event *ev;
	u32 head, reaped = 0;

	head = ring->head;
	do {
		barrier();
		if (head == ring->tail)
			break;
		ev = &ring->events[head];
		if (ev->res != BS) {
			int index = (int) (uintptr_t) ev->obj;
			struct iocb *iocb = &s->sq_ring->iocbs[index];

			printf("io: unexpected ret=%ld\n", ev->res);
			printf("offset=%lu, size=%lu\n", (unsigned long) iocb->u.c.offset, (unsigned long) iocb->u.c.nbytes);
			return -1;
		}
		reaped++;
		head++;
		if (head == ring->nr_events)
			head = 0;
	} while (1);

	s->inflight -= reaped;
	ring->head = head;
	barrier();
	return reaped;
}

static void *submitter_fn(void *data)
{
	struct submitter *s = data;
	int fd, ret, prepped, flags;

	printf("submitter=%d\n", gettid());

	flags = O_RDONLY;
	if (!buffered)
		flags |= O_DIRECT;
	fd = open(s->filename, flags);
	if (fd < 0) {
		perror("open");
		goto done;
	}

	if (get_file_size(fd, &s->max_blocks)) {
		printf("failed getting size of device/file\n");
		goto err;
	}
	if (!s->max_blocks) {
		printf("Zero file/device size?\n");
		goto err;
	}

	s->max_blocks--;

	srand48_r(pthread_self(), &s->rand);

	prepped = 0;
	do {
		int to_wait, flags, to_submit, this_reap;

		if (!prepped && s->inflight < DEPTH)
			prepped = prep_more_ios(s, fd, min(DEPTH - s->inflight, BATCH_SUBMIT));
		s->inflight += prepped;
submit_more:
		to_submit = prepped;
submit:
		if (s->inflight + BATCH_SUBMIT < DEPTH)
			to_wait = 0;
		else
			to_wait = min(s->inflight + to_submit, BATCH_COMPLETE);

		flags = IORING_FLAG_GETEVENTS;
		if (to_submit)
			flags |= IORING_FLAG_SUBMIT;

		ret = io_ring_enter(s->ioc, to_submit, to_wait, flags);
		s->calls++;

		this_reap = reap_events(s);
		if (this_reap == -1)
			break;
		s->reaps += this_reap;

		if (ret >= 0) {
			if (!ret) {
				to_submit = 0;
				if (s->inflight)
					goto submit;
				continue;
			} else if (ret < to_submit) {
				int diff = to_submit - ret;

				s->done += ret;
				prepped -= diff;
				goto submit_more;
			}
			s->done += ret;
			prepped = 0;
			continue;
		} else if (ret < 0) {
			if ((ret == -1 && errno == EAGAIN) || ret == -EAGAIN) {
				if (s->finish)
					break;
				if (this_reap)
					goto submit;
				printf("EAGAIN reap=%d\n", reap_events(s));
				printf("%d/%d, %d/%d\n", s->sq_ring->head, s->sq_ring->tail, s->cq_ring->head, s->cq_ring->tail);
				to_submit = 0;
				goto submit;
			}
			if (ret == -1)
				printf("io_submit: %s\n", strerror(errno));
			else
				printf("io_submit: %s\n", strerror(-ret));
			break;
		}
	} while (!s->finish);
err:
	close(fd);
done:
	finish = 1;
	return NULL;
}

static void sig_int(int sig)
{
	printf("Exiting on signal %d\n", sig);
	submitters[0].finish = 1;
	finish = 1;
}

static void arm_sig_int(void)
{
	struct sigaction act;

	memset(&act, 0, sizeof(act));
	act.sa_handler = sig_int;
	act.sa_flags = SA_RESTART;
	sigaction(SIGINT, &act, NULL);
}

int main(int argc, char *argv[])
{
	struct submitter *s = &submitters[0];
	unsigned long done, calls, reap;
	int flags = 0, err;
	int j;
	size_t size;
	void *p, *ret;
	struct rlimit rlim;

	if (argc < 2) {
		printf("%s: filename\n", argv[0]);
		return 1;
	}

	rlim.rlim_cur = RLIM_INFINITY;
	rlim.rlim_max = RLIM_INFINITY;
	if (setrlimit(RLIMIT_MEMLOCK, &rlim) < 0) {
		perror("setrlimit");
		return 1;
	}

	arm_sig_int();

	size = sizeof(struct aio_iocb_ring) + RING_SIZE * sizeof(struct iocb);
	if (posix_memalign(&p, 4096, size))
		return 1;
	s->sq_ring = p;
	memset(p, 0, size);

	size = sizeof(struct aio_io_event_ring) + RING_SIZE * sizeof(struct io_event);
	if (posix_memalign(&p, 4096, size))
		return 1;
	s->cq_ring = p;
	memset(p, 0, size);

	for (j = 0; j < RING_SIZE; j++) {
		struct iocb *iocb = &s->sq_ring->iocbs[j];

		if (posix_memalign(&iocb->u.c.buf, BS, BS)) {
			printf("failed alloc\n");
			return 1;
		}
		iocb->u.c.nbytes = BS;
	}

	flags = IOCTX_FLAG_SCQRING;
	if (polled)
		flags |= IOCTX_FLAG_IOPOLL;
	if (fixedbufs)
		flags |= IOCTX_FLAG_FIXEDBUFS;
	if (buffered)
		flags |= IOCTX_FLAG_SQWQ;
	else if (sq_thread) {
		flags |= IOCTX_FLAG_SQTHREAD;
		s->sq_ring->sq_thread_cpu = sq_thread_cpu;
	}

	err = io_setup2(RING_SIZE, flags, s->sq_ring->iocbs, s->sq_ring, s->cq_ring, &s->ioc);
	if (err) {
		printf("ctx_init failed: %s, %d\n", strerror(errno), err);
		return 1;
	}
	printf("polled=%d, fixedbufs=%d, buffered=%d\n", polled, fixedbufs, buffered);
	printf("  QD=%d, sq_ring=%d, cq_ring=%d\n", DEPTH, s->sq_ring->nr_events, s->cq_ring->nr_events);
	strcpy(s->filename, argv[1]);

	pthread_create(&s->thread, NULL, submitter_fn, s);

	reap = calls = done = 0;
	do {
		unsigned long this_done = 0;
		unsigned long this_reap = 0;
		unsigned long this_call = 0;
		unsigned long rpc = 0, ipc = 0;

		sleep(1);
		this_done += s->done;
		this_call += s->calls;
		this_reap += s->reaps;
		if (this_call - calls) {
			rpc = (this_done - done) / (this_call - calls);
			ipc = (this_reap - reap) / (this_call - calls);
		}
		printf("IOPS=%lu, IOS/call=%lu/%lu, inflight=%u (head=%d tail=%d), %lu, %lu\n",
				this_done - done, rpc, ipc, s->inflight,
				s->cq_ring->head, s->cq_ring->tail, s->reaps, s->done);
		done = this_done;
		calls = this_call;
		reap = this_reap;
	} while (!finish);

	pthread_join(s->thread, &ret);
	return 0;
}
Commit	Line	Data
c9fb4c5b JA	1	/*
	2	* gcc -D_GNU_SOURCE -Wall -O2 -o aio-ring aio-ring.c -lpthread -laio
	3	*/
	4	#include <stdio.h>
	5	#include <errno.h>
	6	#include <assert.h>
	7	#include <stdlib.h>
	8	#include <stddef.h>
	9	#include <signal.h>
	10	#include <inttypes.h>
	11
	12	#include <sys/types.h>
	13	#include <sys/stat.h>
	14	#include <sys/ioctl.h>
	15	#include <sys/syscall.h>
	16	#include <sys/resource.h>
	17	#include <linux/fs.h>
	18	#include <fcntl.h>
	19	#include <unistd.h>
	20	#include <libaio.h>
	21	#include <string.h>
	22	#include <pthread.h>
	23	#include <sched.h>
	24
	25	#define IOCB_FLAG_HIPRI (1 << 2)
	26
	27	#define IOCTX_FLAG_IOPOLL (1 << 1)
	28	#define IOCTX_FLAG_USERIOCB (1 << 0)
	29	#define IOCTX_FLAG_FIXEDBUFS (1 << 2)
	30	#define IOCTX_FLAG_SCQRING (1 << 3) /* Use SQ/CQ rings */
	31	#define IOCTX_FLAG_SQTHREAD (1 << 4) /* Use SQ thread */
	32	#define IOCTX_FLAG_SQWQ (1 << 5) /* Use SQ wq */
	33
	34	#define barrier() __asm__ __volatile__("": : :"memory")
	35
	36	#define min(a, b) ((a < b) ? (a) : (b))
	37
	38	typedef uint32_t u32;
	39
	40	struct aio_iocb_ring {
	41	union {
	42	struct {
	43	u32 head, tail;
	44	u32 nr_events;
	45	u32 sq_thread_cpu;
	46	};
	47	struct iocb pad_iocb;
	48	};
	49	struct iocb iocbs[0];
	50	};
	51
	52	struct aio_io_event_ring {
	53	union {
	54	struct {
	55	u32 head, tail;
	56	u32 nr_events;
	57	};
	58	struct io_event pad_event;
	59	};
	60	struct io_event events[0];
	61	};
	62
	63	#define IORING_FLAG_SUBMIT (1 << 0)
	64	#define IORING_FLAG_GETEVENTS (1 << 1)
65
66	#define container_of(ptr, type, member) ({ \
67	const __typeof__( ((type )0)->member ) __mptr = (ptr); \
68	(type )( (char )__mptr - offsetof(type,member) );})
69
70	#define DEPTH 32
71	#define RING_SIZE (DEPTH + 1)
72
73	#define BATCH_SUBMIT 8
74	#define BATCH_COMPLETE 8
75
76	#define BS 4096
77
78	struct submitter {
79	pthread_t thread;
80	unsigned long max_blocks;
81	io_context_t ioc;
82	struct drand48_data rand;
83	struct aio_iocb_ring *sq_ring;
84	struct aio_io_event_ring *cq_ring;
85	int inflight;
86	unsigned long reaps;
87	unsigned long done;
88	unsigned long calls;
89	volatile int finish;
90	char filename[128];
91	};
92
93	static struct submitter submitters[1];
94	static volatile int finish;
95
96	static int polled = 1; /* use IO polling */
97	static int fixedbufs = 1; /* use fixed user buffers */
98	static int buffered = 0; /* use buffered IO, not O_DIRECT */
99	static int sq_thread = 0;
100	static int sq_thread_cpu = 0;
101
102	static int io_setup2(unsigned int nr_events, unsigned int flags,
103	struct iocb iocbs, struct aio_iocb_ring sq_ring,
104	struct aio_io_event_ring cq_ring, io_context_t ctx_idp)
105	{
106	return syscall(335, nr_events, flags, iocbs, sq_ring, cq_ring, ctx_idp);
107	}
108
109	static int io_ring_enter(io_context_t ctx, unsigned int to_submit,
110	unsigned int min_complete, unsigned int flags)
111	{
112	return syscall(336, ctx, to_submit, min_complete, flags);
113	}
114
115	static int gettid(void)
116	{
117	return syscall(__NR_gettid);
118	}
119
120	static void init_io(struct submitter s, int fd, struct iocb iocb)
121	{
122	unsigned long offset;
123	long r;
124
125	lrand48_r(&s->rand, &r);
126	offset = (r % (s->max_blocks - 1)) * BS;
127
128	iocb->aio_fildes = fd;
129	iocb->aio_lio_opcode = IO_CMD_PREAD;
130	iocb->u.c.offset = offset;
131	if (polled)
132	iocb->u.c.flags = IOCB_FLAG_HIPRI;
133	if (!fixedbufs)
134	iocb->u.c.nbytes = BS;
135	}
136
137	static int prep_more_ios(struct submitter *s, int fd, int max_ios)
138	{
139	struct aio_iocb_ring *ring = s->sq_ring;
140	struct iocb *iocb;
141	u32 tail, next_tail, prepped = 0;
142
143	next_tail = tail = ring->tail;
144	do {
145	next_tail++;
146	if (next_tail == ring->nr_events)
147	next_tail = 0;
148
149	barrier();
150	if (next_tail == ring->head)
151	break;
152
153	iocb = &s->sq_ring->iocbs[tail];
154	init_io(s, fd, iocb);
155	prepped++;
156	tail = next_tail;
157	} while (prepped < max_ios);
158
159	if (ring->tail != tail) {
160	/* order tail store with writes to iocbs above */
161	barrier();
162	ring->tail = tail;
163	barrier();
164	}
165	return prepped;
166	}
167
168	static int get_file_size(int fd, unsigned long *blocks)
169	{
170	struct stat st;
171
172	if (fstat(fd, &st) < 0)
173	return -1;
174	if (S_ISBLK(st.st_mode)) {
175	unsigned long long bytes;
176
177	if (ioctl(fd, BLKGETSIZE64, &bytes) != 0)
178	return -1;
179
180	*blocks = bytes / BS;
181	return 0;
182	} else if (S_ISREG(st.st_mode)) {
183	*blocks = st.st_size / BS;
184	return 0;
185	}
186
187	return -1;
188	}
189
190	static int reap_events(struct submitter *s)
191	{
192	struct aio_io_event_ring *ring = s->cq_ring;
193	struct io_event *ev;
194	u32 head, reaped = 0;
195
196	head = ring->head;
197	do {
198	barrier();
199	if (head == ring->tail)
200	break;
201	ev = &ring->events[head];
202	if (ev->res != BS) {
203	int index = (int) (uintptr_t) ev->obj;
204	struct iocb *iocb = &s->sq_ring->iocbs[index];
205
206	printf("io: unexpected ret=%ld\n", ev->res);
207	printf("offset=%lu, size=%lu\n", (unsigned long) iocb->u.c.offset, (unsigned long) iocb->u.c.nbytes);
208	return -1;
209	}
210	reaped++;
211	head++;
212	if (head == ring->nr_events)
213	head = 0;
214	} while (1);
215
216	s->inflight -= reaped;
217	ring->head = head;
218	barrier();
219	return reaped;
220	}
221
222	static void submitter_fn(void data)
223	{
224	struct submitter *s = data;
225	int fd, ret, prepped, flags;
226
227	printf("submitter=%d\n", gettid());
228
229	flags = O_RDONLY;
230	if (!buffered)
231	flags \|= O_DIRECT;
232	fd = open(s->filename, flags);
233	if (fd < 0) {
234	perror("open");
235	goto done;
236	}
237
238	if (get_file_size(fd, &s->max_blocks)) {
239	printf("failed getting size of device/file\n");
240	goto err;
241	}
242	if (!s->max_blocks) {
243	printf("Zero file/device size?\n");
244	goto err;
245	}
246
247	s->max_blocks--;
248
249	srand48_r(pthread_self(), &s->rand);
250
251	prepped = 0;
252	do {
253	int to_wait, flags, to_submit, this_reap;
254
255	if (!prepped && s->inflight < DEPTH)
256	prepped = prep_more_ios(s, fd, min(DEPTH - s->inflight, BATCH_SUBMIT));
257	s->inflight += prepped;
258	submit_more:
259	to_submit = prepped;
260	submit:
261	if (s->inflight + BATCH_SUBMIT < DEPTH)
262	to_wait = 0;
263	else
264	to_wait = min(s->inflight + to_submit, BATCH_COMPLETE);
265
266	flags = IORING_FLAG_GETEVENTS;
267	if (to_submit)
268	flags \|= IORING_FLAG_SUBMIT;
269
270	ret = io_ring_enter(s->ioc, to_submit, to_wait, flags);
271	s->calls++;
272
273	this_reap = reap_events(s);
274	if (this_reap == -1)
275	break;
276	s->reaps += this_reap;
277
278	if (ret >= 0) {
279	if (!ret) {
280	to_submit = 0;
281	if (s->inflight)
282	goto submit;
283	continue;
284	} else if (ret < to_submit) {
285	int diff = to_submit - ret;
286
287	s->done += ret;
288	prepped -= diff;
289	goto submit_more;
290	}
291	s->done += ret;
292	prepped = 0;
293	continue;
294	} else if (ret < 0) {
295	if ((ret == -1 && errno == EAGAIN) \|\| ret == -EAGAIN) {
296	if (s->finish)
297	break;
298	if (this_reap)
299	goto submit;
300	printf("EAGAIN reap=%d\n", reap_events(s));
301	printf("%d/%d, %d/%d\n", s->sq_ring->head, s->sq_ring->tail, s->cq_ring->head, s->cq_ring->tail);
302	to_submit = 0;
303	goto submit;
304	}
305	if (ret == -1)
306	printf("io_submit: %s\n", strerror(errno));
307	else
308	printf("io_submit: %s\n", strerror(-ret));
309	break;
310	}
311	} while (!s->finish);
312	err:
313	close(fd);
314	done:
315	finish = 1;
316	return NULL;
317	}
318
319	static void sig_int(int sig)
320	{
321	printf("Exiting on signal %d\n", sig);
322	submitters[0].finish = 1;
323	finish = 1;
324	}
325
326	static void arm_sig_int(void)
327	{
328	struct sigaction act;
329
330	memset(&act, 0, sizeof(act));
331	act.sa_handler = sig_int;
332	act.sa_flags = SA_RESTART;
333	sigaction(SIGINT, &act, NULL);
334	}
335
336	int main(int argc, char *argv[])
337	{
338	struct submitter *s = &submitters[0];
339	unsigned long done, calls, reap;
340	int flags = 0, err;
341	int j;
342	size_t size;
343	void p, ret;
344	struct rlimit rlim;
345
346	if (argc < 2) {
347	printf("%s: filename\n", argv[0]);
348	return 1;
349	}
350
351	rlim.rlim_cur = RLIM_INFINITY;
352	rlim.rlim_max = RLIM_INFINITY;
353	if (setrlimit(RLIMIT_MEMLOCK, &rlim) < 0) {
354	perror("setrlimit");
355	return 1;
356	}
357
358	arm_sig_int();
359
360	size = sizeof(struct aio_iocb_ring) + RING_SIZE * sizeof(struct iocb);
361	if (posix_memalign(&p, 4096, size))
362	return 1;
363	s->sq_ring = p;
364	memset(p, 0, size);
365
366	size = sizeof(struct aio_io_event_ring) + RING_SIZE * sizeof(struct io_event);
367	if (posix_memalign(&p, 4096, size))
368	return 1;
369	s->cq_ring = p;
370	memset(p, 0, size);
371
372	for (j = 0; j < RING_SIZE; j++) {
373	struct iocb *iocb = &s->sq_ring->iocbs[j];
374
375	if (posix_memalign(&iocb->u.c.buf, BS, BS)) {
376	printf("failed alloc\n");
377	return 1;
378	}
379	iocb->u.c.nbytes = BS;
380	}
381
382	flags = IOCTX_FLAG_SCQRING;
383	if (polled)
384	flags \|= IOCTX_FLAG_IOPOLL;
385	if (fixedbufs)
386	flags \|= IOCTX_FLAG_FIXEDBUFS;
387	if (buffered)
388	flags \|= IOCTX_FLAG_SQWQ;
389	else if (sq_thread) {
390	flags \|= IOCTX_FLAG_SQTHREAD;
391	s->sq_ring->sq_thread_cpu = sq_thread_cpu;
392	}
393
394	err = io_setup2(RING_SIZE, flags, s->sq_ring->iocbs, s->sq_ring, s->cq_ring, &s->ioc);
395	if (err) {
396	printf("ctx_init failed: %s, %d\n", strerror(errno), err);
397	return 1;
398	}
399	printf("polled=%d, fixedbufs=%d, buffered=%d\n", polled, fixedbufs, buffered);
400	printf(" QD=%d, sq_ring=%d, cq_ring=%d\n", DEPTH, s->sq_ring->nr_events, s->cq_ring->nr_events);
401	strcpy(s->filename, argv[1]);
402
403	pthread_create(&s->thread, NULL, submitter_fn, s);
404
405	reap = calls = done = 0;
406	do {
407	unsigned long this_done = 0;
408	unsigned long this_reap = 0;
409	unsigned long this_call = 0;
410	unsigned long rpc = 0, ipc = 0;
411
412	sleep(1);
413	this_done += s->done;
414	this_call += s->calls;
415	this_reap += s->reaps;
416	if (this_call - calls) {
417	rpc = (this_done - done) / (this_call - calls);
418	ipc = (this_reap - reap) / (this_call - calls);
419	}
420	printf("IOPS=%lu, IOS/call=%lu/%lu, inflight=%u (head=%d tail=%d), %lu, %lu\n",
421	this_done - done, rpc, ipc, s->inflight,
422	s->cq_ring->head, s->cq_ring->tail, s->reaps, s->done);
423	done = this_done;
424	calls = this_call;
425	reap = this_reap;
426	} while (!finish);
427
428	pthread_join(s->thread, &ret);
429	return 0;
430	}