[fio.git] / t / io_uring.c

#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 <sys/mman.h>
#include <sys/uio.h>
#include <linux/fs.h>
#include <fcntl.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <sched.h>

#include "../arch/arch.h"
#include "../lib/types.h"
#include "../os/io_uring.h"

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

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

struct io_sq_ring {
	unsigned *head;
	unsigned *tail;
	unsigned *ring_mask;
	unsigned *ring_entries;
	unsigned *flags;
	unsigned *array;
};

struct io_cq_ring {
	unsigned *head;
	unsigned *tail;
	unsigned *ring_mask;
	unsigned *ring_entries;
	struct io_uring_cqe *cqes;
};

#define DEPTH			128

#define BATCH_SUBMIT		64
#define BATCH_COMPLETE		64

#define BS			4096

#define MAX_FDS			16

static unsigned sq_ring_mask, cq_ring_mask;

struct file {
	unsigned long max_blocks;
	unsigned pending_ios;
	int real_fd;
	int fixed_fd;
};

struct submitter {
	pthread_t thread;
	int ring_fd;
	struct drand48_data rand;
	struct io_sq_ring sq_ring;
	struct io_uring_sqe *sqes;
	struct iovec iovecs[DEPTH];
	struct io_cq_ring cq_ring;
	int inflight;
	unsigned long reaps;
	unsigned long done;
	unsigned long calls;
	unsigned long cachehit, cachemiss;
	volatile int finish;

	__s32 *fds;

	struct file files[MAX_FDS];
	unsigned nr_files;
	unsigned cur_file;
};

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_poll = 0;	/* use kernel submission/poller thread */
static int sq_thread_cpu = -1;	/* pin above thread to this CPU */
static int do_nop = 0;		/* no-op SQ ring commands */

static int io_uring_register_buffers(struct submitter *s)
{
	struct io_uring_register_buffers reg = {
		.iovecs = s->iovecs,
		.nr_iovecs = DEPTH
	};

	if (do_nop)
		return 0;

	return syscall(__NR_sys_io_uring_register, s->ring_fd,
			IORING_REGISTER_BUFFERS, &reg);
}

static int io_uring_register_files(struct submitter *s)
{
	struct io_uring_register_files reg;
	int i;

	if (do_nop)
		return 0;

	s->fds = calloc(s->nr_files, sizeof(__s32));
	for (i = 0; i < s->nr_files; i++) {
		s->fds[i] = s->files[i].real_fd;
		s->files[i].fixed_fd = i;
	}
	reg.fds = s->fds;
	reg.nr_fds = s->nr_files;

	return syscall(__NR_sys_io_uring_register, s->ring_fd,
			IORING_REGISTER_FILES, &reg);
}

static int io_uring_setup(unsigned entries, struct io_uring_params *p)
{
	return syscall(__NR_sys_io_uring_setup, entries, p);
}

static int io_uring_enter(struct submitter *s, unsigned int to_submit,
			  unsigned int min_complete, unsigned int flags)
{
	return syscall(__NR_sys_io_uring_enter, s->ring_fd, to_submit,
			min_complete, flags);
}

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

static unsigned file_depth(struct submitter *s)
{
	return (DEPTH + s->nr_files - 1) / s->nr_files;
}

static void init_io(struct submitter *s, unsigned index)
{
	struct io_uring_sqe *sqe = &s->sqes[index];
	unsigned long offset;
	struct file *f;
	long r;

	if (do_nop) {
		sqe->opcode = IORING_OP_NOP;
		return;
	}

	if (s->nr_files == 1) {
		f = &s->files[0];
	} else {
		f = &s->files[s->cur_file];
		if (f->pending_ios >= file_depth(s)) {
			s->cur_file++;
			if (s->cur_file == s->nr_files)
				s->cur_file = 0;
		}
	}
	f->pending_ios++;

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

	sqe->flags = IOSQE_FIXED_FILE;
	if (fixedbufs) {
		sqe->opcode = IORING_OP_READ_FIXED;
		sqe->addr = s->iovecs[index].iov_base;
		sqe->len = BS;
		sqe->buf_index = index;
	} else {
		sqe->opcode = IORING_OP_READV;
		sqe->addr = &s->iovecs[index];
		sqe->len = 1;
		sqe->buf_index = 0;
	}
	sqe->ioprio = 0;
	sqe->fd = f->fixed_fd;
	sqe->off = offset;
	sqe->user_data = (unsigned long) f;
}

static int prep_more_ios(struct submitter *s, int max_ios)
{
	struct io_sq_ring *ring = &s->sq_ring;
	unsigned index, tail, next_tail, prepped = 0;

	next_tail = tail = *ring->tail;
	do {
		next_tail++;
		barrier();
		if (next_tail == *ring->head)
			break;

		index = tail & sq_ring_mask;
		init_io(s, index);
		ring->array[index] = index;
		prepped++;
		tail = next_tail;
	} while (prepped < max_ios);

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

static int get_file_size(struct file *f)
{
	struct stat st;

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

		if (ioctl(f->real_fd, BLKGETSIZE64, &bytes) != 0)
			return -1;

		f->max_blocks = bytes / BS;
		return 0;
	} else if (S_ISREG(st.st_mode)) {
		f->max_blocks = st.st_size / BS;
		return 0;
	}

	return -1;
}

static int reap_events(struct submitter *s)
{
	struct io_cq_ring *ring = &s->cq_ring;
	struct io_uring_cqe *cqe;
	unsigned head, reaped = 0;

	head = *ring->head;
	do {
		struct file *f;

		barrier();
		if (head == *ring->tail)
			break;
		cqe = &ring->cqes[head & cq_ring_mask];
		if (!do_nop) {
			f = (struct file *) cqe->user_data;
			f->pending_ios--;
			if (cqe->res != BS) {
				printf("io: unexpected ret=%d\n", cqe->res);
				return -1;
			}
		}
		if (cqe->flags & IOCQE_FLAG_CACHEHIT)
			s->cachehit++;
		else
			s->cachemiss++;
		reaped++;
		head++;
	} while (1);

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

static void *submitter_fn(void *data)
{
	struct submitter *s = data;
	struct io_sq_ring *ring = &s->sq_ring;
	int ret, prepped;

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

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

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

		if (!prepped && s->inflight < DEPTH) {
			to_prep = min(DEPTH - s->inflight, BATCH_SUBMIT);
			prepped = prep_more_ios(s, to_prep);
		}
		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);

		/*
		 * Only need to call io_uring_enter if we're not using SQ thread
		 * poll, or if IORING_SQ_NEED_WAKEUP is set.
		 */
		if (!sq_thread_poll || (*ring->flags & IORING_SQ_NEED_WAKEUP)) {
			unsigned flags = 0;

			if (to_wait)
				flags = IORING_ENTER_GETEVENTS;
			ret = io_uring_enter(s, to_submit, to_wait, flags);
			s->calls++;
		}

		/*
		 * For non SQ thread poll, we already got the events we needed
		 * through the io_uring_enter() above. For SQ thread poll, we
		 * need to loop here until we find enough events.
		 */
		this_reap = 0;
		do {
			int r;
			r = reap_events(s);
			if (r == -1)
				break;
			else if (r > 0)
				this_reap += r;
		} while (sq_thread_poll && this_reap < to_wait);
		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 (errno == EAGAIN) {
				if (s->finish)
					break;
				if (this_reap)
					goto submit;
				to_submit = 0;
				goto submit;
			}
			printf("io_submit: %s\n", strerror(errno));
			break;
		}
	} while (!s->finish);

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

static int setup_ring(struct submitter *s)
{
	struct io_sq_ring *sring = &s->sq_ring;
	struct io_cq_ring *cring = &s->cq_ring;
	struct io_uring_params p;
	int ret, fd;
	void *ptr;

	memset(&p, 0, sizeof(p));

	if (polled)
		p.flags |= IORING_SETUP_IOPOLL;
	if (sq_thread_poll) {
		p.flags |= IORING_SETUP_SQPOLL;
		if (sq_thread_cpu != -1) {
			p.flags |= IORING_SETUP_SQ_AFF;
			p.sq_thread_cpu = sq_thread_cpu;
		}
	}

	fd = io_uring_setup(DEPTH, &p);
	if (fd < 0) {
		perror("io_uring_setup");
		return 1;
	}
	s->ring_fd = fd;

	if (fixedbufs) {
		ret = io_uring_register_buffers(s);
		if (ret < 0) {
			perror("io_uring_register_buffers");
			return 1;
		}
	}

	ret = io_uring_register_files(s);
	if (ret < 0) {
		perror("io_uring_register_files");
		return 1;
	}

	ptr = mmap(0, p.sq_off.array + p.sq_entries * sizeof(__u32),
			PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd,
			IORING_OFF_SQ_RING);
	printf("sq_ring ptr = 0x%p\n", ptr);
	sring->head = ptr + p.sq_off.head;
	sring->tail = ptr + p.sq_off.tail;
	sring->ring_mask = ptr + p.sq_off.ring_mask;
	sring->ring_entries = ptr + p.sq_off.ring_entries;
	sring->flags = ptr + p.sq_off.flags;
	sring->array = ptr + p.sq_off.array;
	sq_ring_mask = *sring->ring_mask;

	s->sqes = mmap(0, p.sq_entries * sizeof(struct io_uring_sqe),
			PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd,
			IORING_OFF_SQES);
	printf("sqes ptr    = 0x%p\n", s->sqes);

	ptr = mmap(0, p.cq_off.cqes + p.cq_entries * sizeof(struct io_uring_cqe),
			PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd,
			IORING_OFF_CQ_RING);
	printf("cq_ring ptr = 0x%p\n", ptr);
	cring->head = ptr + p.cq_off.head;
	cring->tail = ptr + p.cq_off.tail;
	cring->ring_mask = ptr + p.cq_off.ring_mask;
	cring->ring_entries = ptr + p.cq_off.ring_entries;
	cring->cqes = ptr + p.cq_off.cqes;
	cq_ring_mask = *cring->ring_mask;
	return 0;
}

int main(int argc, char *argv[])
{
	struct submitter *s = &submitters[0];
	unsigned long done, calls, reap, cache_hit, cache_miss;
	int err, i, flags, fd;
	struct rlimit rlim;
	void *ret;

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

	flags = O_RDONLY | O_NOATIME;
	if (!buffered)
		flags |= O_DIRECT;

	i = 1;
	while (!do_nop && i < argc) {
		struct file *f = &s->files[s->nr_files];

		fd = open(argv[i], flags);
		if (fd < 0) {
			perror("open");
			return 1;
		}
		f->real_fd = fd;
		if (get_file_size(f)) {
			printf("failed getting size of device/file\n");
			return 1;
		}
		if (f->max_blocks <= 1) {
			printf("Zero file/device size?\n");
			return 1;
		}
		f->max_blocks--;

		printf("Added file %s\n", argv[i]);
		s->nr_files++;
		i++;
	}

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

	arm_sig_int();

	for (i = 0; i < DEPTH; i++) {
		void *buf;

		if (posix_memalign(&buf, BS, BS)) {
			printf("failed alloc\n");
			return 1;
		}
		s->iovecs[i].iov_base = buf;
		s->iovecs[i].iov_len = BS;
	}

	err = setup_ring(s);
	if (err) {
		printf("ring setup failed: %s, %d\n", strerror(errno), err);
		return 1;
	}
	printf("polled=%d, fixedbufs=%d, buffered=%d", polled, fixedbufs, buffered);
	printf(" QD=%d, sq_ring=%d, cq_ring=%d\n", DEPTH, *s->sq_ring.ring_entries, *s->cq_ring.ring_entries);

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

	cache_hit = cache_miss = reap = calls = done = 0;
	do {
		unsigned long this_done = 0;
		unsigned long this_reap = 0;
		unsigned long this_call = 0;
		unsigned long this_cache_hit = 0;
		unsigned long this_cache_miss = 0;
		unsigned long rpc = 0, ipc = 0;
		double hit = 0.0;

		sleep(1);
		this_done += s->done;
		this_call += s->calls;
		this_reap += s->reaps;
		this_cache_hit += s->cachehit;
		this_cache_miss += s->cachemiss;
		if (this_cache_hit && this_cache_miss) {
			unsigned long hits, total;

			hits = this_cache_hit - cache_hit;
			total = hits + this_cache_miss - cache_miss;
			hit = (double) hits / (double) total;
			hit *= 100.0;
		}
		if (this_call - calls) {
			rpc = (this_done - done) / (this_call - calls);
			ipc = (this_reap - reap) / (this_call - calls);
		} else
			rpc = ipc = -1;
		printf("IOPS=%lu, IOS/call=%ld/%ld, inflight=%u (head=%u tail=%u), Cachehit=%0.2f%%\n",
				this_done - done, rpc, ipc, s->inflight,
				*s->cq_ring.head, *s->cq_ring.tail, hit);
		done = this_done;
		calls = this_call;
		reap = this_reap;
		cache_hit = s->cachehit;
		cache_miss = s->cachemiss;
	} while (!finish);

	pthread_join(s->thread, &ret);
	close(s->ring_fd);
	return 0;
}
Commit	Line	Data
	1	#include <stdio.h>
	2	#include <errno.h>
	3	#include <assert.h>
	4	#include <stdlib.h>
	5	#include <stddef.h>
	6	#include <signal.h>
	7	#include <inttypes.h>
	8
	9	#include <sys/types.h>
	10	#include <sys/stat.h>
	11	#include <sys/ioctl.h>
	12	#include <sys/syscall.h>
	13	#include <sys/resource.h>
	14	#include <sys/mman.h>
	15	#include <sys/uio.h>
	16	#include <linux/fs.h>
	17	#include <fcntl.h>
	18	#include <unistd.h>
	19	#include <string.h>
	20	#include <pthread.h>
	21	#include <sched.h>
	22
	23	#include "../arch/arch.h"
	24	#include "../lib/types.h"
	25	#include "../os/io_uring.h"
	26
	27	#define barrier() __asm__ __volatile__("": : :"memory")
	28
	29	#define min(a, b) ((a < b) ? (a) : (b))
	30
	31	struct io_sq_ring {
	32	unsigned *head;
	33	unsigned *tail;
	34	unsigned *ring_mask;
	35	unsigned *ring_entries;
	36	unsigned *flags;
	37	unsigned *array;
	38	};
	39
	40	struct io_cq_ring {
	41	unsigned *head;
	42	unsigned *tail;
	43	unsigned *ring_mask;
	44	unsigned *ring_entries;
	45	struct io_uring_cqe *cqes;
	46	};
	47
	48	#define DEPTH 128
	49
	50	#define BATCH_SUBMIT 64
	51	#define BATCH_COMPLETE 64
	52
	53	#define BS 4096
	54
	55	#define MAX_FDS 16
	56
	57	static unsigned sq_ring_mask, cq_ring_mask;
	58
	59	struct file {
	60	unsigned long max_blocks;
	61	unsigned pending_ios;
	62	int real_fd;
	63	int fixed_fd;
	64	};
	65
	66	struct submitter {
	67	pthread_t thread;
	68	int ring_fd;
	69	struct drand48_data rand;
	70	struct io_sq_ring sq_ring;
	71	struct io_uring_sqe *sqes;
	72	struct iovec iovecs[DEPTH];
	73	struct io_cq_ring cq_ring;
	74	int inflight;
	75	unsigned long reaps;
	76	unsigned long done;
	77	unsigned long calls;
	78	unsigned long cachehit, cachemiss;
	79	volatile int finish;
	80
	81	__s32 *fds;
	82
	83	struct file files[MAX_FDS];
	84	unsigned nr_files;
	85	unsigned cur_file;
	86	};
	87
	88	static struct submitter submitters[1];
	89	static volatile int finish;
	90
	91	static int polled = 1; /* use IO polling */
	92	static int fixedbufs = 1; /* use fixed user buffers */
	93	static int buffered = 0; /* use buffered IO, not O_DIRECT */
	94	static int sq_thread_poll = 0; /* use kernel submission/poller thread */
	95	static int sq_thread_cpu = -1; /* pin above thread to this CPU */
	96	static int do_nop = 0; /* no-op SQ ring commands */
	97
	98	static int io_uring_register_buffers(struct submitter *s)
	99	{
	100	struct io_uring_register_buffers reg = {
	101	.iovecs = s->iovecs,
	102	.nr_iovecs = DEPTH
	103	};
	104
	105	if (do_nop)
	106	return 0;
	107
	108	return syscall(__NR_sys_io_uring_register, s->ring_fd,
	109	IORING_REGISTER_BUFFERS, &reg);
	110	}
	111
	112	static int io_uring_register_files(struct submitter *s)
	113	{
	114	struct io_uring_register_files reg;
	115	int i;
	116
	117	if (do_nop)
	118	return 0;
	119
	120	s->fds = calloc(s->nr_files, sizeof(__s32));
	121	for (i = 0; i < s->nr_files; i++) {
	122	s->fds[i] = s->files[i].real_fd;
	123	s->files[i].fixed_fd = i;
	124	}
	125	reg.fds = s->fds;
	126	reg.nr_fds = s->nr_files;
	127
	128	return syscall(__NR_sys_io_uring_register, s->ring_fd,
	129	IORING_REGISTER_FILES, &reg);
	130	}
	131
	132	static int io_uring_setup(unsigned entries, struct io_uring_params *p)
	133	{
	134	return syscall(__NR_sys_io_uring_setup, entries, p);
	135	}
	136
	137	static int io_uring_enter(struct submitter *s, unsigned int to_submit,
	138	unsigned int min_complete, unsigned int flags)
	139	{
	140	return syscall(__NR_sys_io_uring_enter, s->ring_fd, to_submit,
	141	min_complete, flags);
	142	}
	143
	144	static int gettid(void)
	145	{
	146	return syscall(__NR_gettid);
	147	}
	148
	149	static unsigned file_depth(struct submitter *s)
	150	{
	151	return (DEPTH + s->nr_files - 1) / s->nr_files;
	152	}
	153
	154	static void init_io(struct submitter *s, unsigned index)
	155	{
	156	struct io_uring_sqe *sqe = &s->sqes[index];
	157	unsigned long offset;
	158	struct file *f;
	159	long r;
	160
	161	if (do_nop) {
	162	sqe->opcode = IORING_OP_NOP;
	163	return;
	164	}
	165
	166	if (s->nr_files == 1) {
	167	f = &s->files[0];
	168	} else {
	169	f = &s->files[s->cur_file];
	170	if (f->pending_ios >= file_depth(s)) {
	171	s->cur_file++;
	172	if (s->cur_file == s->nr_files)
	173	s->cur_file = 0;
	174	}
	175	}
	176	f->pending_ios++;
	177
	178	lrand48_r(&s->rand, &r);
	179	offset = (r % (f->max_blocks - 1)) * BS;
	180
	181	sqe->flags = IOSQE_FIXED_FILE;
	182	if (fixedbufs) {
	183	sqe->opcode = IORING_OP_READ_FIXED;
	184	sqe->addr = s->iovecs[index].iov_base;
	185	sqe->len = BS;
	186	sqe->buf_index = index;
	187	} else {
	188	sqe->opcode = IORING_OP_READV;
	189	sqe->addr = &s->iovecs[index];
	190	sqe->len = 1;
	191	sqe->buf_index = 0;
	192	}
	193	sqe->ioprio = 0;
	194	sqe->fd = f->fixed_fd;
	195	sqe->off = offset;
	196	sqe->user_data = (unsigned long) f;
	197	}
	198
	199	static int prep_more_ios(struct submitter *s, int max_ios)
	200	{
	201	struct io_sq_ring *ring = &s->sq_ring;
	202	unsigned index, tail, next_tail, prepped = 0;
	203
	204	next_tail = tail = *ring->tail;
	205	do {
	206	next_tail++;
	207	barrier();
	208	if (next_tail == *ring->head)
	209	break;
	210
	211	index = tail & sq_ring_mask;
	212	init_io(s, index);
	213	ring->array[index] = index;
	214	prepped++;
	215	tail = next_tail;
	216	} while (prepped < max_ios);
	217
	218	if (*ring->tail != tail) {
	219	/* order tail store with writes to sqes above */
	220	barrier();
	221	*ring->tail = tail;
	222	barrier();
	223	}
	224	return prepped;
	225	}
	226
	227	static int get_file_size(struct file *f)
	228	{
	229	struct stat st;
	230
	231	if (fstat(f->real_fd, &st) < 0)
	232	return -1;
	233	if (S_ISBLK(st.st_mode)) {
	234	unsigned long long bytes;
	235
	236	if (ioctl(f->real_fd, BLKGETSIZE64, &bytes) != 0)
	237	return -1;
	238
	239	f->max_blocks = bytes / BS;
	240	return 0;
	241	} else if (S_ISREG(st.st_mode)) {
	242	f->max_blocks = st.st_size / BS;
	243	return 0;
	244	}
	245
	246	return -1;
	247	}
	248
	249	static int reap_events(struct submitter *s)
	250	{
	251	struct io_cq_ring *ring = &s->cq_ring;
	252	struct io_uring_cqe *cqe;
	253	unsigned head, reaped = 0;
	254
	255	head = *ring->head;
	256	do {
	257	struct file *f;
	258
	259	barrier();
	260	if (head == *ring->tail)
	261	break;
	262	cqe = &ring->cqes[head & cq_ring_mask];
	263	if (!do_nop) {
	264	f = (struct file *) cqe->user_data;
	265	f->pending_ios--;
	266	if (cqe->res != BS) {
	267	printf("io: unexpected ret=%d\n", cqe->res);
	268	return -1;
	269	}
	270	}
	271	if (cqe->flags & IOCQE_FLAG_CACHEHIT)
	272	s->cachehit++;
	273	else
	274	s->cachemiss++;
	275	reaped++;
	276	head++;
	277	} while (1);
	278
	279	s->inflight -= reaped;
	280	*ring->head = head;
	281	barrier();
	282	return reaped;
	283	}
	284
	285	static void submitter_fn(void data)
	286	{
	287	struct submitter *s = data;
	288	struct io_sq_ring *ring = &s->sq_ring;
	289	int ret, prepped;
	290
	291	printf("submitter=%d\n", gettid());
	292
	293	srand48_r(pthread_self(), &s->rand);
	294
	295	prepped = 0;
	296	do {
	297	int to_wait, to_submit, this_reap, to_prep;
	298
	299	if (!prepped && s->inflight < DEPTH) {
	300	to_prep = min(DEPTH - s->inflight, BATCH_SUBMIT);
	301	prepped = prep_more_ios(s, to_prep);
	302	}
	303	s->inflight += prepped;
	304	submit_more:
	305	to_submit = prepped;
	306	submit:
	307	if (s->inflight + BATCH_SUBMIT < DEPTH)
	308	to_wait = 0;
	309	else
	310	to_wait = min(s->inflight + to_submit, BATCH_COMPLETE);
	311
	312	/*
	313	* Only need to call io_uring_enter if we're not using SQ thread
	314	* poll, or if IORING_SQ_NEED_WAKEUP is set.
	315	*/
	316	if (!sq_thread_poll \|\| (*ring->flags & IORING_SQ_NEED_WAKEUP)) {
	317	unsigned flags = 0;
	318
	319	if (to_wait)
	320	flags = IORING_ENTER_GETEVENTS;
	321	ret = io_uring_enter(s, to_submit, to_wait, flags);
	322	s->calls++;
	323	}
	324
	325	/*
	326	* For non SQ thread poll, we already got the events we needed
	327	* through the io_uring_enter() above. For SQ thread poll, we
	328	* need to loop here until we find enough events.
	329	*/
	330	this_reap = 0;
	331	do {
	332	int r;
	333	r = reap_events(s);
	334	if (r == -1)
	335	break;
	336	else if (r > 0)
	337	this_reap += r;
	338	} while (sq_thread_poll && this_reap < to_wait);
	339	s->reaps += this_reap;
	340
	341	if (ret >= 0) {
	342	if (!ret) {
	343	to_submit = 0;
	344	if (s->inflight)
	345	goto submit;
	346	continue;
	347	} else if (ret < to_submit) {
	348	int diff = to_submit - ret;
	349
	350	s->done += ret;
	351	prepped -= diff;
	352	goto submit_more;
	353	}
	354	s->done += ret;
	355	prepped = 0;
	356	continue;
	357	} else if (ret < 0) {
	358	if (errno == EAGAIN) {
	359	if (s->finish)
	360	break;
	361	if (this_reap)
	362	goto submit;
	363	to_submit = 0;
	364	goto submit;
	365	}
	366	printf("io_submit: %s\n", strerror(errno));
	367	break;
	368	}
	369	} while (!s->finish);
	370
	371	finish = 1;
	372	return NULL;
	373	}
	374
	375	static void sig_int(int sig)
	376	{
	377	printf("Exiting on signal %d\n", sig);
	378	submitters[0].finish = 1;
	379	finish = 1;
	380	}
	381
	382	static void arm_sig_int(void)
	383	{
	384	struct sigaction act;
	385
	386	memset(&act, 0, sizeof(act));
	387	act.sa_handler = sig_int;
	388	act.sa_flags = SA_RESTART;
	389	sigaction(SIGINT, &act, NULL);
	390	}
	391
	392	static int setup_ring(struct submitter *s)
	393	{
	394	struct io_sq_ring *sring = &s->sq_ring;
	395	struct io_cq_ring *cring = &s->cq_ring;
	396	struct io_uring_params p;
	397	int ret, fd;
	398	void *ptr;
	399
	400	memset(&p, 0, sizeof(p));
	401
	402	if (polled)
	403	p.flags \|= IORING_SETUP_IOPOLL;
	404	if (sq_thread_poll) {
	405	p.flags \|= IORING_SETUP_SQPOLL;
	406	if (sq_thread_cpu != -1) {
	407	p.flags \|= IORING_SETUP_SQ_AFF;
	408	p.sq_thread_cpu = sq_thread_cpu;
	409	}
	410	}
	411
	412	fd = io_uring_setup(DEPTH, &p);
	413	if (fd < 0) {
	414	perror("io_uring_setup");
	415	return 1;
	416	}
	417	s->ring_fd = fd;
	418
	419	if (fixedbufs) {
	420	ret = io_uring_register_buffers(s);
	421	if (ret < 0) {
	422	perror("io_uring_register_buffers");
	423	return 1;
	424	}
	425	}
	426
	427	ret = io_uring_register_files(s);
	428	if (ret < 0) {
	429	perror("io_uring_register_files");
	430	return 1;
	431	}
	432
	433	ptr = mmap(0, p.sq_off.array + p.sq_entries * sizeof(__u32),
	434	PROT_READ \| PROT_WRITE, MAP_SHARED \| MAP_POPULATE, fd,
	435	IORING_OFF_SQ_RING);
	436	printf("sq_ring ptr = 0x%p\n", ptr);
	437	sring->head = ptr + p.sq_off.head;
	438	sring->tail = ptr + p.sq_off.tail;
	439	sring->ring_mask = ptr + p.sq_off.ring_mask;
	440	sring->ring_entries = ptr + p.sq_off.ring_entries;
	441	sring->flags = ptr + p.sq_off.flags;
	442	sring->array = ptr + p.sq_off.array;
	443	sq_ring_mask = *sring->ring_mask;
	444
	445	s->sqes = mmap(0, p.sq_entries * sizeof(struct io_uring_sqe),
	446	PROT_READ \| PROT_WRITE, MAP_SHARED \| MAP_POPULATE, fd,
	447	IORING_OFF_SQES);
	448	printf("sqes ptr = 0x%p\n", s->sqes);
	449
	450	ptr = mmap(0, p.cq_off.cqes + p.cq_entries * sizeof(struct io_uring_cqe),
	451	PROT_READ \| PROT_WRITE, MAP_SHARED \| MAP_POPULATE, fd,
	452	IORING_OFF_CQ_RING);
	453	printf("cq_ring ptr = 0x%p\n", ptr);
	454	cring->head = ptr + p.cq_off.head;
	455	cring->tail = ptr + p.cq_off.tail;
	456	cring->ring_mask = ptr + p.cq_off.ring_mask;
	457	cring->ring_entries = ptr + p.cq_off.ring_entries;
	458	cring->cqes = ptr + p.cq_off.cqes;
	459	cq_ring_mask = *cring->ring_mask;
	460	return 0;
	461	}
	462
	463	int main(int argc, char *argv[])
	464	{
	465	struct submitter *s = &submitters[0];
	466	unsigned long done, calls, reap, cache_hit, cache_miss;
	467	int err, i, flags, fd;
	468	struct rlimit rlim;
	469	void *ret;
	470
	471	if (!do_nop && argc < 2) {
	472	printf("%s: filename\n", argv[0]);
	473	return 1;
	474	}
	475
	476	flags = O_RDONLY \| O_NOATIME;
	477	if (!buffered)
	478	flags \|= O_DIRECT;
	479
	480	i = 1;
	481	while (!do_nop && i < argc) {
	482	struct file *f = &s->files[s->nr_files];
	483
	484	fd = open(argv[i], flags);
	485	if (fd < 0) {
	486	perror("open");
	487	return 1;
	488	}
	489	f->real_fd = fd;
	490	if (get_file_size(f)) {
	491	printf("failed getting size of device/file\n");
	492	return 1;
	493	}
	494	if (f->max_blocks <= 1) {
	495	printf("Zero file/device size?\n");
	496	return 1;
	497	}
	498	f->max_blocks--;
	499
	500	printf("Added file %s\n", argv[i]);
	501	s->nr_files++;
	502	i++;
	503	}
	504
	505	rlim.rlim_cur = RLIM_INFINITY;
	506	rlim.rlim_max = RLIM_INFINITY;
	507	if (setrlimit(RLIMIT_MEMLOCK, &rlim) < 0) {
	508	perror("setrlimit");
	509	return 1;
	510	}
	511
	512	arm_sig_int();
	513
	514	for (i = 0; i < DEPTH; i++) {
	515	void *buf;
	516
	517	if (posix_memalign(&buf, BS, BS)) {
	518	printf("failed alloc\n");
	519	return 1;
	520	}
	521	s->iovecs[i].iov_base = buf;
	522	s->iovecs[i].iov_len = BS;
	523	}
	524
	525	err = setup_ring(s);
	526	if (err) {
	527	printf("ring setup failed: %s, %d\n", strerror(errno), err);
	528	return 1;
	529	}
	530	printf("polled=%d, fixedbufs=%d, buffered=%d", polled, fixedbufs, buffered);
	531	printf(" QD=%d, sq_ring=%d, cq_ring=%d\n", DEPTH, s->sq_ring.ring_entries, s->cq_ring.ring_entries);
	532
	533	pthread_create(&s->thread, NULL, submitter_fn, s);
	534
	535	cache_hit = cache_miss = reap = calls = done = 0;
	536	do {
	537	unsigned long this_done = 0;
	538	unsigned long this_reap = 0;
	539	unsigned long this_call = 0;
	540	unsigned long this_cache_hit = 0;
	541	unsigned long this_cache_miss = 0;
	542	unsigned long rpc = 0, ipc = 0;
	543	double hit = 0.0;
	544
	545	sleep(1);
	546	this_done += s->done;
	547	this_call += s->calls;
	548	this_reap += s->reaps;
	549	this_cache_hit += s->cachehit;
	550	this_cache_miss += s->cachemiss;
	551	if (this_cache_hit && this_cache_miss) {
	552	unsigned long hits, total;
	553
	554	hits = this_cache_hit - cache_hit;
	555	total = hits + this_cache_miss - cache_miss;
	556	hit = (double) hits / (double) total;
	557	hit *= 100.0;
	558	}
	559	if (this_call - calls) {
	560	rpc = (this_done - done) / (this_call - calls);
	561	ipc = (this_reap - reap) / (this_call - calls);
	562	} else
	563	rpc = ipc = -1;
	564	printf("IOPS=%lu, IOS/call=%ld/%ld, inflight=%u (head=%u tail=%u), Cachehit=%0.2f%%\n",
	565	this_done - done, rpc, ipc, s->inflight,
	566	s->cq_ring.head, s->cq_ring.tail, hit);
	567	done = this_done;
	568	calls = this_call;
	569	reap = this_reap;
	570	cache_hit = s->cachehit;
	571	cache_miss = s->cachemiss;
	572	} while (!finish);
	573
	574	pthread_join(s->thread, &ret);
	575	close(s->ring_fd);
	576	return 0;
	577	}