[qemu.git] / block / io_uring.c

/*
 * Linux io_uring support.
 *
 * Copyright (C) 2009 IBM, Corp.
 * Copyright (C) 2009 Red Hat, Inc.
 * Copyright (C) 2019 Aarushi Mehta
 *
 * This work is licensed under the terms of the GNU GPL, version 2 or later.
 * See the COPYING file in the top-level directory.
 */
#include "qemu/osdep.h"
#include <liburing.h>
#include "qemu-common.h"
#include "block/aio.h"
#include "qemu/queue.h"
#include "block/block.h"
#include "block/raw-aio.h"
#include "qemu/coroutine.h"
#include "qapi/error.h"
#include "trace.h"

/* io_uring ring size */
#define MAX_ENTRIES 128

typedef struct LuringAIOCB {
    Coroutine *co;
    struct io_uring_sqe sqeq;
    ssize_t ret;
    QEMUIOVector *qiov;
    bool is_read;
    QSIMPLEQ_ENTRY(LuringAIOCB) next;

    /*
     * Buffered reads may require resubmission, see
     * luring_resubmit_short_read().
     */
    int total_read;
    QEMUIOVector resubmit_qiov;
} LuringAIOCB;

typedef struct LuringQueue {
    int plugged;
    unsigned int in_queue;
    unsigned int in_flight;
    bool blocked;
    QSIMPLEQ_HEAD(, LuringAIOCB) submit_queue;
} LuringQueue;

typedef struct LuringState {
    AioContext *aio_context;

    struct io_uring ring;

    /* io queue for submit at batch.  Protected by AioContext lock. */
    LuringQueue io_q;

    /* I/O completion processing.  Only runs in I/O thread.  */
    QEMUBH *completion_bh;
} LuringState;

/**
 * luring_resubmit:
 *
 * Resubmit a request by appending it to submit_queue.  The caller must ensure
 * that ioq_submit() is called later so that submit_queue requests are started.
 */
static void luring_resubmit(LuringState *s, LuringAIOCB *luringcb)
{
    QSIMPLEQ_INSERT_TAIL(&s->io_q.submit_queue, luringcb, next);
    s->io_q.in_queue++;
}

/**
 * luring_resubmit_short_read:
 *
 * Before Linux commit 9d93a3f5a0c ("io_uring: punt short reads to async
 * context") a buffered I/O request with the start of the file range in the
 * page cache could result in a short read.  Applications need to resubmit the
 * remaining read request.
 *
 * This is a slow path but recent kernels never take it.
 */
static void luring_resubmit_short_read(LuringState *s, LuringAIOCB *luringcb,
                                       int nread)
{
    QEMUIOVector *resubmit_qiov;
    size_t remaining;

    trace_luring_resubmit_short_read(s, luringcb, nread);

    /* Update read position */
    luringcb->total_read = nread;
    remaining = luringcb->qiov->size - luringcb->total_read;

    /* Shorten qiov */
    resubmit_qiov = &luringcb->resubmit_qiov;
    if (resubmit_qiov->iov == NULL) {
        qemu_iovec_init(resubmit_qiov, luringcb->qiov->niov);
    } else {
        qemu_iovec_reset(resubmit_qiov);
    }
    qemu_iovec_concat(resubmit_qiov, luringcb->qiov, luringcb->total_read,
                      remaining);

    /* Update sqe */
    luringcb->sqeq.off = nread;
    luringcb->sqeq.addr = (__u64)(uintptr_t)luringcb->resubmit_qiov.iov;
    luringcb->sqeq.len = luringcb->resubmit_qiov.niov;

    luring_resubmit(s, luringcb);
}

/**
 * luring_process_completions:
 * @s: AIO state
 *
 * Fetches completed I/O requests, consumes cqes and invokes their callbacks
 * The function is somewhat tricky because it supports nested event loops, for
 * example when a request callback invokes aio_poll().
 *
 * Function schedules BH completion so it  can be called again in a nested
 * event loop.  When there are no events left  to complete the BH is being
 * canceled.
 *
 */
static void luring_process_completions(LuringState *s)
{
    struct io_uring_cqe *cqes;
    int total_bytes;
    /*
     * Request completion callbacks can run the nested event loop.
     * Schedule ourselves so the nested event loop will "see" remaining
     * completed requests and process them.  Without this, completion
     * callbacks that wait for other requests using a nested event loop
     * would hang forever.
     *
     * This workaround is needed because io_uring uses poll_wait, which
     * is woken up when new events are added to the uring, thus polling on
     * the same uring fd will block unless more events are received.
     *
     * Other leaf block drivers (drivers that access the data themselves)
     * are networking based, so they poll sockets for data and run the
     * correct coroutine.
     */
    qemu_bh_schedule(s->completion_bh);

    while (io_uring_peek_cqe(&s->ring, &cqes) == 0) {
        LuringAIOCB *luringcb;
        int ret;

        if (!cqes) {
            break;
        }

        luringcb = io_uring_cqe_get_data(cqes);
        ret = cqes->res;
        io_uring_cqe_seen(&s->ring, cqes);
        cqes = NULL;

        /* Change counters one-by-one because we can be nested. */
        s->io_q.in_flight--;
        trace_luring_process_completion(s, luringcb, ret);

        /* total_read is non-zero only for resubmitted read requests */
        total_bytes = ret + luringcb->total_read;

        if (ret < 0) {
            if (ret == -EINTR) {
                luring_resubmit(s, luringcb);
                continue;
            }
        } else if (!luringcb->qiov) {
            goto end;
        } else if (total_bytes == luringcb->qiov->size) {
            ret = 0;
        /* Only read/write */
        } else {
            /* Short Read/Write */
            if (luringcb->is_read) {
                if (ret > 0) {
                    luring_resubmit_short_read(s, luringcb, ret);
                    continue;
                } else {
                    /* Pad with zeroes */
                    qemu_iovec_memset(luringcb->qiov, total_bytes, 0,
                                      luringcb->qiov->size - total_bytes);
                    ret = 0;
                }
            } else {
                ret = -ENOSPC;
            }
        }
end:
        luringcb->ret = ret;
        qemu_iovec_destroy(&luringcb->resubmit_qiov);

        /*
         * If the coroutine is already entered it must be in ioq_submit()
         * and will notice luringcb->ret has been filled in when it
         * eventually runs later. Coroutines cannot be entered recursively
         * so avoid doing that!
         */
        if (!qemu_coroutine_entered(luringcb->co)) {
            aio_co_wake(luringcb->co);
        }
    }
    qemu_bh_cancel(s->completion_bh);
}

static int ioq_submit(LuringState *s)
{
    int ret = 0;
    LuringAIOCB *luringcb, *luringcb_next;

    while (s->io_q.in_queue > 0) {
        /*
         * Try to fetch sqes from the ring for requests waiting in
         * the overflow queue
         */
        QSIMPLEQ_FOREACH_SAFE(luringcb, &s->io_q.submit_queue, next,
                              luringcb_next) {
            struct io_uring_sqe *sqes = io_uring_get_sqe(&s->ring);
            if (!sqes) {
                break;
            }
            /* Prep sqe for submission */
            *sqes = luringcb->sqeq;
            QSIMPLEQ_REMOVE_HEAD(&s->io_q.submit_queue, next);
        }
        ret = io_uring_submit(&s->ring);
        trace_luring_io_uring_submit(s, ret);
        /* Prevent infinite loop if submission is refused */
        if (ret <= 0) {
            if (ret == -EAGAIN) {
                continue;
            }
            break;
        }
        s->io_q.in_flight += ret;
        s->io_q.in_queue  -= ret;
    }
    s->io_q.blocked = (s->io_q.in_queue > 0);

    if (s->io_q.in_flight) {
        /*
         * We can try to complete something just right away if there are
         * still requests in-flight.
         */
        luring_process_completions(s);
    }
    return ret;
}

static void luring_process_completions_and_submit(LuringState *s)
{
    aio_context_acquire(s->aio_context);
    luring_process_completions(s);

    if (!s->io_q.plugged && s->io_q.in_queue > 0) {
        ioq_submit(s);
    }
    aio_context_release(s->aio_context);
}

static void qemu_luring_completion_bh(void *opaque)
{
    LuringState *s = opaque;
    luring_process_completions_and_submit(s);
}

static void qemu_luring_completion_cb(void *opaque)
{
    LuringState *s = opaque;
    luring_process_completions_and_submit(s);
}

static bool qemu_luring_poll_cb(void *opaque)
{
    LuringState *s = opaque;
    struct io_uring_cqe *cqes;

    if (io_uring_peek_cqe(&s->ring, &cqes) == 0) {
        if (cqes) {
            luring_process_completions_and_submit(s);
            return true;
        }
    }

    return false;
}

static void ioq_init(LuringQueue *io_q)
{
    QSIMPLEQ_INIT(&io_q->submit_queue);
    io_q->plugged = 0;
    io_q->in_queue = 0;
    io_q->in_flight = 0;
    io_q->blocked = false;
}

void luring_io_plug(BlockDriverState *bs, LuringState *s)
{
    trace_luring_io_plug(s);
    s->io_q.plugged++;
}

void luring_io_unplug(BlockDriverState *bs, LuringState *s)
{
    assert(s->io_q.plugged);
    trace_luring_io_unplug(s, s->io_q.blocked, s->io_q.plugged,
                           s->io_q.in_queue, s->io_q.in_flight);
    if (--s->io_q.plugged == 0 &&
        !s->io_q.blocked && s->io_q.in_queue > 0) {
        ioq_submit(s);
    }
}

/**
 * luring_do_submit:
 * @fd: file descriptor for I/O
 * @luringcb: AIO control block
 * @s: AIO state
 * @offset: offset for request
 * @type: type of request
 *
 * Fetches sqes from ring, adds to pending queue and preps them
 *
 */
static int luring_do_submit(int fd, LuringAIOCB *luringcb, LuringState *s,
                            uint64_t offset, int type)
{
    int ret;
    struct io_uring_sqe *sqes = &luringcb->sqeq;

    switch (type) {
    case QEMU_AIO_WRITE:
        io_uring_prep_writev(sqes, fd, luringcb->qiov->iov,
                             luringcb->qiov->niov, offset);
        break;
    case QEMU_AIO_READ:
        io_uring_prep_readv(sqes, fd, luringcb->qiov->iov,
                            luringcb->qiov->niov, offset);
        break;
    case QEMU_AIO_FLUSH:
        io_uring_prep_fsync(sqes, fd, IORING_FSYNC_DATASYNC);
        break;
    default:
        fprintf(stderr, "%s: invalid AIO request type, aborting 0x%x.\n",
                        __func__, type);
        abort();
    }
    io_uring_sqe_set_data(sqes, luringcb);

    QSIMPLEQ_INSERT_TAIL(&s->io_q.submit_queue, luringcb, next);
    s->io_q.in_queue++;
    trace_luring_do_submit(s, s->io_q.blocked, s->io_q.plugged,
                           s->io_q.in_queue, s->io_q.in_flight);
    if (!s->io_q.blocked &&
        (!s->io_q.plugged ||
         s->io_q.in_flight + s->io_q.in_queue >= MAX_ENTRIES)) {
        ret = ioq_submit(s);
        trace_luring_do_submit_done(s, ret);
        return ret;
    }
    return 0;
}

int coroutine_fn luring_co_submit(BlockDriverState *bs, LuringState *s, int fd,
                                  uint64_t offset, QEMUIOVector *qiov, int type)
{
    int ret;
    LuringAIOCB luringcb = {
        .co         = qemu_coroutine_self(),
        .ret        = -EINPROGRESS,
        .qiov       = qiov,
        .is_read    = (type == QEMU_AIO_READ),
    };
    trace_luring_co_submit(bs, s, &luringcb, fd, offset, qiov ? qiov->size : 0,
                           type);
    ret = luring_do_submit(fd, &luringcb, s, offset, type);

    if (ret < 0) {
        return ret;
    }

    if (luringcb.ret == -EINPROGRESS) {
        qemu_coroutine_yield();
    }
    return luringcb.ret;
}

void luring_detach_aio_context(LuringState *s, AioContext *old_context)
{
    aio_set_fd_handler(old_context, s->ring.ring_fd, false, NULL, NULL, NULL,
                       s);
    qemu_bh_delete(s->completion_bh);
    s->aio_context = NULL;
}

void luring_attach_aio_context(LuringState *s, AioContext *new_context)
{
    s->aio_context = new_context;
    s->completion_bh = aio_bh_new(new_context, qemu_luring_completion_bh, s);
    aio_set_fd_handler(s->aio_context, s->ring.ring_fd, false,
                       qemu_luring_completion_cb, NULL, qemu_luring_poll_cb, s);
}

LuringState *luring_init(Error **errp)
{
    int rc;
    LuringState *s = g_new0(LuringState, 1);
    struct io_uring *ring = &s->ring;

    trace_luring_init_state(s, sizeof(*s));

    rc = io_uring_queue_init(MAX_ENTRIES, ring, 0);
    if (rc < 0) {
        error_setg_errno(errp, errno, "failed to init linux io_uring ring");
        g_free(s);
        return NULL;
    }

    ioq_init(&s->io_q);
    return s;

}

void luring_cleanup(LuringState *s)
{
    io_uring_queue_exit(&s->ring);
    g_free(s);
    trace_luring_cleanup_state(s);
}
Commit	Line	Data
6663a0a3 AM	1	/*
	2	* Linux io_uring support.
	3	*
	4	* Copyright (C) 2009 IBM, Corp.
	5	* Copyright (C) 2009 Red Hat, Inc.
	6	* Copyright (C) 2019 Aarushi Mehta
	7	*
	8	* This work is licensed under the terms of the GNU GPL, version 2 or later.
	9	* See the COPYING file in the top-level directory.
	10	*/
	11	#include "qemu/osdep.h"
	12	#include <liburing.h>
	13	#include "qemu-common.h"
	14	#include "block/aio.h"
	15	#include "qemu/queue.h"
	16	#include "block/block.h"
	17	#include "block/raw-aio.h"
	18	#include "qemu/coroutine.h"
	19	#include "qapi/error.h"
d803f590	20	#include "trace.h"
6663a0a3 AM	21
	22	/* io_uring ring size */
	23	#define MAX_ENTRIES 128
	24
	25	typedef struct LuringAIOCB {
	26	Coroutine *co;
	27	struct io_uring_sqe sqeq;
	28	ssize_t ret;
	29	QEMUIOVector *qiov;
	30	bool is_read;
	31	QSIMPLEQ_ENTRY(LuringAIOCB) next;
	32
	33	/*
	34	* Buffered reads may require resubmission, see
	35	* luring_resubmit_short_read().
	36	*/
	37	int total_read;
	38	QEMUIOVector resubmit_qiov;
	39	} LuringAIOCB;
	40
	41	typedef struct LuringQueue {
	42	int plugged;
	43	unsigned int in_queue;
	44	unsigned int in_flight;
	45	bool blocked;
	46	QSIMPLEQ_HEAD(, LuringAIOCB) submit_queue;
	47	} LuringQueue;
	48
	49	typedef struct LuringState {
	50	AioContext *aio_context;
	51
	52	struct io_uring ring;
	53
	54	/* io queue for submit at batch. Protected by AioContext lock. */
	55	LuringQueue io_q;
	56
	57	/* I/O completion processing. Only runs in I/O thread. */
	58	QEMUBH *completion_bh;
	59	} LuringState;
	60
	61	/**
	62	* luring_resubmit:
	63	*
	64	* Resubmit a request by appending it to submit_queue. The caller must ensure
	65	* that ioq_submit() is called later so that submit_queue requests are started.
	66	*/
	67	static void luring_resubmit(LuringState s, LuringAIOCB luringcb)
	68	{
	69	QSIMPLEQ_INSERT_TAIL(&s->io_q.submit_queue, luringcb, next);
	70	s->io_q.in_queue++;
	71	}
	72
	73	/**
	74	* luring_resubmit_short_read:
	75	*
	76	* Before Linux commit 9d93a3f5a0c ("io_uring: punt short reads to async
	77	* context") a buffered I/O request with the start of the file range in the
	78	* page cache could result in a short read. Applications need to resubmit the
	79	* remaining read request.
	80	*
	81	* This is a slow path but recent kernels never take it.
	82	*/
	83	static void luring_resubmit_short_read(LuringState s, LuringAIOCB luringcb,
	84	int nread)
85	{
86	QEMUIOVector *resubmit_qiov;
87	size_t remaining;
88
d803f590 AM	89	trace_luring_resubmit_short_read(s, luringcb, nread);
d803f590 AM	90
6663a0a3 AM	91	/* Update read position */
	92	luringcb->total_read = nread;
	93	remaining = luringcb->qiov->size - luringcb->total_read;
	94
	95	/* Shorten qiov */
	96	resubmit_qiov = &luringcb->resubmit_qiov;
	97	if (resubmit_qiov->iov == NULL) {
	98	qemu_iovec_init(resubmit_qiov, luringcb->qiov->niov);
	99	} else {
	100	qemu_iovec_reset(resubmit_qiov);
	101	}
	102	qemu_iovec_concat(resubmit_qiov, luringcb->qiov, luringcb->total_read,
	103	remaining);
	104
	105	/* Update sqe */
	106	luringcb->sqeq.off = nread;
	107	luringcb->sqeq.addr = (__u64)(uintptr_t)luringcb->resubmit_qiov.iov;
	108	luringcb->sqeq.len = luringcb->resubmit_qiov.niov;
	109
	110	luring_resubmit(s, luringcb);
	111	}
	112
	113	/**
	114	* luring_process_completions:
	115	* @s: AIO state
	116	*
	117	* Fetches completed I/O requests, consumes cqes and invokes their callbacks
	118	* The function is somewhat tricky because it supports nested event loops, for
	119	* example when a request callback invokes aio_poll().
	120	*
	121	* Function schedules BH completion so it can be called again in a nested
	122	* event loop. When there are no events left to complete the BH is being
	123	* canceled.
	124	*
	125	*/
	126	static void luring_process_completions(LuringState *s)
	127	{
	128	struct io_uring_cqe *cqes;
	129	int total_bytes;
	130	/*
	131	* Request completion callbacks can run the nested event loop.
	132	* Schedule ourselves so the nested event loop will "see" remaining
	133	* completed requests and process them. Without this, completion
	134	* callbacks that wait for other requests using a nested event loop
	135	* would hang forever.
	136	*
	137	* This workaround is needed because io_uring uses poll_wait, which
	138	* is woken up when new events are added to the uring, thus polling on
	139	* the same uring fd will block unless more events are received.
	140	*
	141	* Other leaf block drivers (drivers that access the data themselves)
	142	* are networking based, so they poll sockets for data and run the
	143	* correct coroutine.
	144	*/
	145	qemu_bh_schedule(s->completion_bh);
	146
	147	while (io_uring_peek_cqe(&s->ring, &cqes) == 0) {
	148	LuringAIOCB *luringcb;
	149	int ret;
	150
	151	if (!cqes) {
	152	break;
	153	}
	154
155	luringcb = io_uring_cqe_get_data(cqes);
156	ret = cqes->res;
157	io_uring_cqe_seen(&s->ring, cqes);
158	cqes = NULL;
159
160	/* Change counters one-by-one because we can be nested. */
161	s->io_q.in_flight--;
d803f590	162	trace_luring_process_completion(s, luringcb, ret);
6663a0a3 AM	163
	164	/* total_read is non-zero only for resubmitted read requests */
	165	total_bytes = ret + luringcb->total_read;
	166
	167	if (ret < 0) {
	168	if (ret == -EINTR) {
	169	luring_resubmit(s, luringcb);
	170	continue;
	171	}
	172	} else if (!luringcb->qiov) {
	173	goto end;
	174	} else if (total_bytes == luringcb->qiov->size) {
	175	ret = 0;
	176	/* Only read/write */
	177	} else {
	178	/* Short Read/Write */
	179	if (luringcb->is_read) {
	180	if (ret > 0) {
	181	luring_resubmit_short_read(s, luringcb, ret);
	182	continue;
	183	} else {
	184	/* Pad with zeroes */
	185	qemu_iovec_memset(luringcb->qiov, total_bytes, 0,
	186	luringcb->qiov->size - total_bytes);
	187	ret = 0;
	188	}
	189	} else {
74e4a8a9	190	ret = -ENOSPC;
6663a0a3 AM	191	}
	192	}
	193	end:
	194	luringcb->ret = ret;
	195	qemu_iovec_destroy(&luringcb->resubmit_qiov);
	196
	197	/*
	198	* If the coroutine is already entered it must be in ioq_submit()
	199	* and will notice luringcb->ret has been filled in when it
	200	* eventually runs later. Coroutines cannot be entered recursively
	201	* so avoid doing that!
	202	*/
	203	if (!qemu_coroutine_entered(luringcb->co)) {
	204	aio_co_wake(luringcb->co);
	205	}
	206	}
	207	qemu_bh_cancel(s->completion_bh);
	208	}
	209
	210	static int ioq_submit(LuringState *s)
	211	{
	212	int ret = 0;
	213	LuringAIOCB luringcb, luringcb_next;
	214
	215	while (s->io_q.in_queue > 0) {
	216	/*
	217	* Try to fetch sqes from the ring for requests waiting in
	218	* the overflow queue
	219	*/
	220	QSIMPLEQ_FOREACH_SAFE(luringcb, &s->io_q.submit_queue, next,
	221	luringcb_next) {
	222	struct io_uring_sqe *sqes = io_uring_get_sqe(&s->ring);
	223	if (!sqes) {
	224	break;
	225	}
	226	/* Prep sqe for submission */
	227	*sqes = luringcb->sqeq;
	228	QSIMPLEQ_REMOVE_HEAD(&s->io_q.submit_queue, next);
	229	}
	230	ret = io_uring_submit(&s->ring);
d803f590	231	trace_luring_io_uring_submit(s, ret);
6663a0a3 AM	232	/* Prevent infinite loop if submission is refused */
	233	if (ret <= 0) {
	234	if (ret == -EAGAIN) {
	235	continue;
	236	}
	237	break;
	238	}
	239	s->io_q.in_flight += ret;
	240	s->io_q.in_queue -= ret;
	241	}
	242	s->io_q.blocked = (s->io_q.in_queue > 0);
	243
	244	if (s->io_q.in_flight) {
	245	/*
	246	* We can try to complete something just right away if there are
	247	* still requests in-flight.
	248	*/
	249	luring_process_completions(s);
	250	}
	251	return ret;
	252	}
	253
	254	static void luring_process_completions_and_submit(LuringState *s)
	255	{
	256	aio_context_acquire(s->aio_context);
	257	luring_process_completions(s);
	258
	259	if (!s->io_q.plugged && s->io_q.in_queue > 0) {
	260	ioq_submit(s);
	261	}
	262	aio_context_release(s->aio_context);
	263	}
	264
	265	static void qemu_luring_completion_bh(void *opaque)
	266	{
	267	LuringState *s = opaque;
	268	luring_process_completions_and_submit(s);
	269	}
	270
	271	static void qemu_luring_completion_cb(void *opaque)
	272	{
	273	LuringState *s = opaque;
	274	luring_process_completions_and_submit(s);
	275	}
	276
daffeb02 AM	277	static bool qemu_luring_poll_cb(void *opaque)
	278	{
	279	LuringState *s = opaque;
	280	struct io_uring_cqe *cqes;
	281
	282	if (io_uring_peek_cqe(&s->ring, &cqes) == 0) {
	283	if (cqes) {
	284	luring_process_completions_and_submit(s);
	285	return true;
	286	}
	287	}
	288
	289	return false;
	290	}
	291
6663a0a3 AM	292	static void ioq_init(LuringQueue *io_q)
	293	{
	294	QSIMPLEQ_INIT(&io_q->submit_queue);
	295	io_q->plugged = 0;
	296	io_q->in_queue = 0;
	297	io_q->in_flight = 0;
	298	io_q->blocked = false;
	299	}
	300
	301	void luring_io_plug(BlockDriverState bs, LuringState s)
	302	{
d803f590	303	trace_luring_io_plug(s);
6663a0a3 AM	304	s->io_q.plugged++;
	305	}
	306
	307	void luring_io_unplug(BlockDriverState bs, LuringState s)
	308	{
	309	assert(s->io_q.plugged);
d803f590 AM	310	trace_luring_io_unplug(s, s->io_q.blocked, s->io_q.plugged,
d803f590 AM	311	s->io_q.in_queue, s->io_q.in_flight);
6663a0a3 AM	312	if (--s->io_q.plugged == 0 &&
	313	!s->io_q.blocked && s->io_q.in_queue > 0) {
	314	ioq_submit(s);
	315	}
	316	}
	317
	318	/**
	319	* luring_do_submit:
	320	* @fd: file descriptor for I/O
	321	* @luringcb: AIO control block
	322	* @s: AIO state
	323	* @offset: offset for request
	324	* @type: type of request
	325	*
	326	* Fetches sqes from ring, adds to pending queue and preps them
	327	*
	328	*/
	329	static int luring_do_submit(int fd, LuringAIOCB luringcb, LuringState s,
	330	uint64_t offset, int type)
	331	{
d803f590	332	int ret;
6663a0a3 AM	333	struct io_uring_sqe *sqes = &luringcb->sqeq;
	334
	335	switch (type) {
	336	case QEMU_AIO_WRITE:
	337	io_uring_prep_writev(sqes, fd, luringcb->qiov->iov,
	338	luringcb->qiov->niov, offset);
	339	break;
	340	case QEMU_AIO_READ:
	341	io_uring_prep_readv(sqes, fd, luringcb->qiov->iov,
	342	luringcb->qiov->niov, offset);
	343	break;
	344	case QEMU_AIO_FLUSH:
	345	io_uring_prep_fsync(sqes, fd, IORING_FSYNC_DATASYNC);
	346	break;
	347	default:
	348	fprintf(stderr, "%s: invalid AIO request type, aborting 0x%x.\n",
	349	__func__, type);
	350	abort();
	351	}
	352	io_uring_sqe_set_data(sqes, luringcb);
	353
	354	QSIMPLEQ_INSERT_TAIL(&s->io_q.submit_queue, luringcb, next);
	355	s->io_q.in_queue++;
d803f590 AM	356	trace_luring_do_submit(s, s->io_q.blocked, s->io_q.plugged,
d803f590 AM	357	s->io_q.in_queue, s->io_q.in_flight);
6663a0a3 AM	358	if (!s->io_q.blocked &&
	359	(!s->io_q.plugged \|\|
	360	s->io_q.in_flight + s->io_q.in_queue >= MAX_ENTRIES)) {
d803f590 AM	361	ret = ioq_submit(s);
	362	trace_luring_do_submit_done(s, ret);
	363	return ret;
6663a0a3 AM	364	}
	365	return 0;
	366	}
	367
	368	int coroutine_fn luring_co_submit(BlockDriverState bs, LuringState s, int fd,
	369	uint64_t offset, QEMUIOVector *qiov, int type)
	370	{
	371	int ret;
	372	LuringAIOCB luringcb = {
	373	.co = qemu_coroutine_self(),
	374	.ret = -EINPROGRESS,
	375	.qiov = qiov,
	376	.is_read = (type == QEMU_AIO_READ),
	377	};
d803f590 AM	378	trace_luring_co_submit(bs, s, &luringcb, fd, offset, qiov ? qiov->size : 0,
d803f590 AM	379	type);
6663a0a3	380	ret = luring_do_submit(fd, &luringcb, s, offset, type);
d803f590	381
6663a0a3 AM	382	if (ret < 0) {
	383	return ret;
	384	}
	385
	386	if (luringcb.ret == -EINPROGRESS) {
	387	qemu_coroutine_yield();
	388	}
	389	return luringcb.ret;
	390	}
	391
	392	void luring_detach_aio_context(LuringState s, AioContext old_context)
	393	{
	394	aio_set_fd_handler(old_context, s->ring.ring_fd, false, NULL, NULL, NULL,
	395	s);
	396	qemu_bh_delete(s->completion_bh);
	397	s->aio_context = NULL;
	398	}
	399
	400	void luring_attach_aio_context(LuringState s, AioContext new_context)
	401	{
	402	s->aio_context = new_context;
	403	s->completion_bh = aio_bh_new(new_context, qemu_luring_completion_bh, s);
	404	aio_set_fd_handler(s->aio_context, s->ring.ring_fd, false,
daffeb02	405	qemu_luring_completion_cb, NULL, qemu_luring_poll_cb, s);
6663a0a3 AM	406	}
	407
	408	LuringState luring_init(Error *errp)
	409	{
	410	int rc;
	411	LuringState *s = g_new0(LuringState, 1);
	412	struct io_uring *ring = &s->ring;
	413
d803f590 AM	414	trace_luring_init_state(s, sizeof(*s));
d803f590 AM	415
6663a0a3 AM	416	rc = io_uring_queue_init(MAX_ENTRIES, ring, 0);
	417	if (rc < 0) {
	418	error_setg_errno(errp, errno, "failed to init linux io_uring ring");
	419	g_free(s);
	420	return NULL;
	421	}
	422
	423	ioq_init(&s->io_q);
	424	return s;
	425
	426	}
	427
	428	void luring_cleanup(LuringState *s)
	429	{
	430	io_uring_queue_exit(&s->ring);
	431	g_free(s);
d803f590	432	trace_luring_cleanup_state(s);
6663a0a3	433	}