[Qemu-devel] [PATCH v3 1/3] linux-aio: fix submit aio as a batch
In the enqueue path, we can't complete request, otherwise "Co-routine re-entered recursively" may be caused, so this patch fixes the issue with below ideas: - for -EAGAIN or partial completion, retry the submision by schedule an BH in following completion cb - for part of completion, also update the io queue - for other failure, return the failure if in enqueue path, otherwise, abort all queued I/O Signed-off-by: Ming Lei --- block/linux-aio.c | 101 + 1 file changed, 79 insertions(+), 22 deletions(-) diff --git a/block/linux-aio.c b/block/linux-aio.c index d92513b..f66e8ad 100644 --- a/block/linux-aio.c +++ b/block/linux-aio.c @@ -38,11 +38,19 @@ struct qemu_laiocb { QLIST_ENTRY(qemu_laiocb) node; }; +/* + * TODO: support to batch I/O from multiple bs in one same + * AIO context, one important use case is multi-lun scsi, + * so in future the IO queue should be per AIO context. + */ typedef struct { struct iocb *iocbs[MAX_QUEUED_IO]; int plugged; unsigned int size; unsigned int idx; + +/* handle -EAGAIN and partial completion */ +QEMUBH *retry; } LaioQueue; struct qemu_laio_state { @@ -137,6 +145,12 @@ static void qemu_laio_completion_bh(void *opaque) } } +static void qemu_laio_start_retry(struct qemu_laio_state *s) +{ +if (s->io_q.idx) +qemu_bh_schedule(s->io_q.retry); +} + static void qemu_laio_completion_cb(EventNotifier *e) { struct qemu_laio_state *s = container_of(e, struct qemu_laio_state, e); @@ -144,6 +158,7 @@ static void qemu_laio_completion_cb(EventNotifier *e) if (event_notifier_test_and_clear(&s->e)) { qemu_bh_schedule(s->completion_bh); } +qemu_laio_start_retry(s); } static void laio_cancel(BlockAIOCB *blockacb) @@ -163,6 +178,9 @@ static void laio_cancel(BlockAIOCB *blockacb) } laiocb->common.cb(laiocb->common.opaque, laiocb->ret); + +/* check if there are requests in io queue */ +qemu_laio_start_retry(laiocb->ctx); } static const AIOCBInfo laio_aiocb_info = { @@ -177,45 +195,80 @@ static void ioq_init(LaioQueue *io_q) io_q->plugged = 0; } -static int ioq_submit(struct qemu_laio_state *s) +static void abort_queue(struct qemu_laio_state *s) +{ +int i; +for (i = 0; i < s->io_q.idx; i++) { +struct qemu_laiocb *laiocb = container_of(s->io_q.iocbs[i], + struct qemu_laiocb, + iocb); +laiocb->ret = -EIO; +qemu_laio_process_completion(s, laiocb); +} +} + +static int ioq_submit(struct qemu_laio_state *s, bool enqueue) { int ret, i = 0; int len = s->io_q.idx; +int j = 0; -do { -ret = io_submit(s->ctx, len, s->io_q.iocbs); -} while (i++ < 3 && ret == -EAGAIN); +if (!len) { +return 0; +} -/* empty io queue */ -s->io_q.idx = 0; +ret = io_submit(s->ctx, len, s->io_q.iocbs); +if (ret == -EAGAIN) { /* retry in following completion cb */ +return 0; +} else if (ret < 0) { +if (enqueue) { +return ret; +} -if (ret < 0) { -i = 0; -} else { -i = ret; +/* in non-queue path, all IOs have to be completed */ +abort_queue(s); +ret = len; +} else if (ret == 0) { +goto out; } -for (; i < len; i++) { -struct qemu_laiocb *laiocb = -container_of(s->io_q.iocbs[i], struct qemu_laiocb, iocb); - -laiocb->ret = (ret < 0) ? ret : -EIO; -qemu_laio_process_completion(s, laiocb); +for (i = ret; i < len; i++) { +s->io_q.iocbs[j++] = s->io_q.iocbs[i]; } + + out: +/* + * update io queue, for partial completion, retry will be + * started automatically in following completion cb. + */ +s->io_q.idx -= ret; + return ret; } -static void ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb) +static void ioq_submit_retry(void *opaque) +{ +struct qemu_laio_state *s = opaque; +ioq_submit(s, false); +} + +static int ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb) { unsigned int idx = s->io_q.idx; +if (unlikely(idx == s->io_q.size)) { +return -1; +} + s->io_q.iocbs[idx++] = iocb; s->io_q.idx = idx; -/* submit immediately if queue is full */ -if (idx == s->io_q.size) { -ioq_submit(s); +/* submit immediately if queue depth is above 2/3 */ +if (idx > s->io_q.size * 2 / 3) { +return ioq_submit(s, true); } + +return 0; } void laio_io_plug(BlockDriverState *bs, void *aio_ctx) @@ -237,7 +290,7 @@ int laio_io_unplug(BlockDriverState *bs, void *aio_ctx, bool unplug) } if (s->io_q.idx > 0) { -ret = ioq_submit(s); +ret = ioq_submit(s, false); } return ret; @@ -281,7 +334,9 @@ BlockAIOCB *laio_submit(BlockD
Re: [Qemu-devel] [PATCH v3 1/3] linux-aio: fix submit aio as a batch
On Tue, Nov 18, 2014 at 10:18 PM, Paolo Bonzini wrote: > > >> @@ -137,6 +145,12 @@ static void qemu_laio_completion_bh(void *opaque) >> } >> } >> >> +static void qemu_laio_start_retry(struct qemu_laio_state *s) >> +{ >> +if (s->io_q.idx) >> +qemu_bh_schedule(s->io_q.retry); >> +} >> + >> static void qemu_laio_completion_cb(EventNotifier *e) >> { >> struct qemu_laio_state *s = container_of(e, struct qemu_laio_state, e); >> @@ -144,6 +158,7 @@ static void qemu_laio_completion_cb(EventNotifier *e) >> if (event_notifier_test_and_clear(&s->e)) { >> qemu_bh_schedule(s->completion_bh); >> } >> +qemu_laio_start_retry(s); > > I think you do not even need two bottom halves. Just call ioq_submit > from completion_bh instead, after the call to io_getevents. Yes, that can save one BH, actually the patch was written when there wasn't completion BH, :-) > >> } >> >> static void laio_cancel(BlockAIOCB *blockacb) >> @@ -163,6 +178,9 @@ static void laio_cancel(BlockAIOCB *blockacb) >> } >> >> laiocb->common.cb(laiocb->common.opaque, laiocb->ret); >> + >> +/* check if there are requests in io queue */ >> +qemu_laio_start_retry(laiocb->ctx); >> } >> >> static const AIOCBInfo laio_aiocb_info = { >> @@ -177,45 +195,80 @@ static void ioq_init(LaioQueue *io_q) >> io_q->plugged = 0; >> } >> >> -static int ioq_submit(struct qemu_laio_state *s) >> +static void abort_queue(struct qemu_laio_state *s) >> +{ >> +int i; >> +for (i = 0; i < s->io_q.idx; i++) { >> +struct qemu_laiocb *laiocb = container_of(s->io_q.iocbs[i], >> + struct qemu_laiocb, >> + iocb); >> +laiocb->ret = -EIO; >> +qemu_laio_process_completion(s, laiocb); >> +} >> +} >> + >> +static int ioq_submit(struct qemu_laio_state *s, bool enqueue) >> { >> int ret, i = 0; >> int len = s->io_q.idx; >> +int j = 0; >> >> -do { >> -ret = io_submit(s->ctx, len, s->io_q.iocbs); >> -} while (i++ < 3 && ret == -EAGAIN); >> +if (!len) { >> +return 0; >> +} >> >> -/* empty io queue */ >> -s->io_q.idx = 0; >> +ret = io_submit(s->ctx, len, s->io_q.iocbs); >> +if (ret == -EAGAIN) { /* retry in following completion cb */ >> +return 0; >> +} else if (ret < 0) { >> +if (enqueue) { >> +return ret; >> +} >> >> -if (ret < 0) { >> -i = 0; >> -} else { >> -i = ret; >> +/* in non-queue path, all IOs have to be completed */ >> +abort_queue(s); >> +ret = len; >> +} else if (ret == 0) { >> +goto out; > > No need for goto; just move the "for" loop inside this conditional. Or > better, just use memmove. That is: > > if (ret < 0) { > if (ret == -EAGAIN) { > return 0; > } > if (enqueue) { > return ret; > } > > abort_queue(s); > ret = len; > } > > if (ret > 0) { > memmove(...) > s->io_q.idx -= ret; > } > return ret; The above is better. >> + * update io queue, for partial completion, retry will be >> + * started automatically in following completion cb. >> + */ >> +s->io_q.idx -= ret; >> + >> return ret; >> } >> >> -static void ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb) >> +static void ioq_submit_retry(void *opaque) >> +{ >> +struct qemu_laio_state *s = opaque; >> +ioq_submit(s, false); >> +} >> + >> +static int ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb) >> { >> unsigned int idx = s->io_q.idx; >> >> +if (unlikely(idx == s->io_q.size)) { >> +return -1; > > return -EAGAIN? It means the io queue is full, so the code has to fail the current request. Thanks, Ming Lei
Re: [Qemu-devel] [PATCH v3 1/3] linux-aio: fix submit aio as a batch
On 22/11/2014 13:16, Ming Lei wrote: > > > +static int ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb) > > > { > > > unsigned int idx = s->io_q.idx; > > > > > > +if (unlikely(idx == s->io_q.size)) { > > > +return -1; > > > > return -EAGAIN? > > It means the io queue is full, so the code has to fail the current > request. Right, but "-1" is "-EPERM" which doesn't make much sense. I think the right thing to do is to return EAGAIN, and let the owner figure it out. For example, a SCSI device might return a "BUSY" status code. Paolo
Re: [Qemu-devel] [PATCH v3 1/3] linux-aio: fix submit aio as a batch
On Mon, Nov 24, 2014 at 5:47 PM, Paolo Bonzini wrote: > > > On 22/11/2014 13:16, Ming Lei wrote: >> > > +static int ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb) >> > > { >> > > unsigned int idx = s->io_q.idx; >> > > >> > > +if (unlikely(idx == s->io_q.size)) { >> > > +return -1; >> > >> > return -EAGAIN? >> >> It means the io queue is full, so the code has to fail the current >> request. > > Right, but "-1" is "-EPERM" which doesn't make much sense. I think the > right thing to do is to return EAGAIN, and let the owner figure it out. > For example, a SCSI device might return a "BUSY" status code. We can do that, but laio_submit() doesn't return the code to callers. Thanks, Ming Lei
Re: [Qemu-devel] [PATCH v3 1/3] linux-aio: fix submit aio as a batch
On 24/11/2014 11:02, Ming Lei wrote: >> > Right, but "-1" is "-EPERM" which doesn't make much sense. I think the >> > right thing to do is to return EAGAIN, and let the owner figure it out. >> > For example, a SCSI device might return a "BUSY" status code. > We can do that, but laio_submit() doesn't return the code to callers. That can be fixed later. But not returning a sensible errno only makes things more complicated later on. Paolo
Re: [Qemu-devel] [PATCH v3 1/3] linux-aio: fix submit aio as a batch
> @@ -137,6 +145,12 @@ static void qemu_laio_completion_bh(void *opaque) > } > } > > +static void qemu_laio_start_retry(struct qemu_laio_state *s) > +{ > +if (s->io_q.idx) > +qemu_bh_schedule(s->io_q.retry); > +} > + > static void qemu_laio_completion_cb(EventNotifier *e) > { > struct qemu_laio_state *s = container_of(e, struct qemu_laio_state, e); > @@ -144,6 +158,7 @@ static void qemu_laio_completion_cb(EventNotifier *e) > if (event_notifier_test_and_clear(&s->e)) { > qemu_bh_schedule(s->completion_bh); > } > +qemu_laio_start_retry(s); I think you do not even need two bottom halves. Just call ioq_submit from completion_bh instead, after the call to io_getevents. > } > > static void laio_cancel(BlockAIOCB *blockacb) > @@ -163,6 +178,9 @@ static void laio_cancel(BlockAIOCB *blockacb) > } > > laiocb->common.cb(laiocb->common.opaque, laiocb->ret); > + > +/* check if there are requests in io queue */ > +qemu_laio_start_retry(laiocb->ctx); > } > > static const AIOCBInfo laio_aiocb_info = { > @@ -177,45 +195,80 @@ static void ioq_init(LaioQueue *io_q) > io_q->plugged = 0; > } > > -static int ioq_submit(struct qemu_laio_state *s) > +static void abort_queue(struct qemu_laio_state *s) > +{ > +int i; > +for (i = 0; i < s->io_q.idx; i++) { > +struct qemu_laiocb *laiocb = container_of(s->io_q.iocbs[i], > + struct qemu_laiocb, > + iocb); > +laiocb->ret = -EIO; > +qemu_laio_process_completion(s, laiocb); > +} > +} > + > +static int ioq_submit(struct qemu_laio_state *s, bool enqueue) > { > int ret, i = 0; > int len = s->io_q.idx; > +int j = 0; > > -do { > -ret = io_submit(s->ctx, len, s->io_q.iocbs); > -} while (i++ < 3 && ret == -EAGAIN); > +if (!len) { > +return 0; > +} > > -/* empty io queue */ > -s->io_q.idx = 0; > +ret = io_submit(s->ctx, len, s->io_q.iocbs); > +if (ret == -EAGAIN) { /* retry in following completion cb */ > +return 0; > +} else if (ret < 0) { > +if (enqueue) { > +return ret; > +} > > -if (ret < 0) { > -i = 0; > -} else { > -i = ret; > +/* in non-queue path, all IOs have to be completed */ > +abort_queue(s); > +ret = len; > +} else if (ret == 0) { > +goto out; No need for goto; just move the "for" loop inside this conditional. Or better, just use memmove. That is: if (ret < 0) { if (ret == -EAGAIN) { return 0; } if (enqueue) { return ret; } abort_queue(s); ret = len; } if (ret > 0) { memmove(...) s->io_q.idx -= ret; } return ret; > + * update io queue, for partial completion, retry will be > + * started automatically in following completion cb. > + */ > +s->io_q.idx -= ret; > + > return ret; > } > > -static void ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb) > +static void ioq_submit_retry(void *opaque) > +{ > +struct qemu_laio_state *s = opaque; > +ioq_submit(s, false); > +} > + > +static int ioq_enqueue(struct qemu_laio_state *s, struct iocb *iocb) > { > unsigned int idx = s->io_q.idx; > > +if (unlikely(idx == s->io_q.size)) { > +return -1; return -EAGAIN? Thanks, Paolo > +} > + > s->io_q.iocbs[idx++] = iocb; > s->io_q.idx = idx; > > -/* submit immediately if queue is full */ > -if (idx == s->io_q.size) { > -ioq_submit(s); > +/* submit immediately if queue depth is above 2/3 */ > +if (idx > s->io_q.size * 2 / 3) { > +return ioq_submit(s, true); > } > + > +return 0; > } > > void laio_io_plug(BlockDriverState *bs, void *aio_ctx) > @@ -237,7 +290,7 @@ int laio_io_unplug(BlockDriverState *bs, void *aio_ctx, > bool unplug) > } > > if (s->io_q.idx > 0) { > -ret = ioq_submit(s); > +ret = ioq_submit(s, false); > } > > return ret; > @@ -281,7 +334,9 @@ BlockAIOCB *laio_submit(BlockDriverState *bs, void > *aio_ctx, int fd, > goto out_free_aiocb; > } > } else { > -ioq_enqueue(s, iocbs); > +if (ioq_enqueue(s, iocbs) < 0) { > +goto out_free_aiocb; > +} > } > return &laiocb->common; > > @@ -296,12 +351,14 @@ void laio_detach_aio_context(void *s_, AioContext > *old_context) > > aio_set_event_notifier(old_context, &s->e, NULL); > qemu_bh_delete(s->completion_bh); > +qemu_bh_delete(s->io_q.retry); > } > > void laio_attach_aio_context(void *s_, AioContext *new_context) > { > struct qemu_laio_state *s = s_; > > +s->io_q.retry = aio_bh_new(new_context, ioq_submit_retry, s); > s->completion_bh = aio_bh_new(new_context, qemu_laio_completion_b