From 3e845ce01a391d7c5d59ff2f28db5381bf02fa27 Mon Sep 17 00:00:00 2001 From: Kent Overstreet Date: Tue, 7 May 2013 16:18:51 -0700 Subject: aio: change reqs_active to include unreaped completions The aio code tries really hard to avoid having to deal with the completion ringbuffer overflowing. To do that, it has to keep track of the number of outstanding kiocbs, and the number of completions currently in the ringbuffer - and it's got to check that every time we allocate a kiocb. Ouch. But - we can improve this quite a bit if we just change reqs_active to mean "number of outstanding requests and unreaped completions" - that means kiocb allocation doesn't have to look at the ringbuffer, which is a fairly significant win. Signed-off-by: Kent Overstreet Cc: Zach Brown Cc: Felipe Balbi Cc: Greg Kroah-Hartman Cc: Mark Fasheh Cc: Joel Becker Cc: Rusty Russell Cc: Jens Axboe Cc: Asai Thambi S P Cc: Selvan Mani Cc: Sam Bradshaw Cc: Jeff Moyer Cc: Al Viro Cc: Benjamin LaHaise Signed-off-by: "Theodore Ts'o" Signed-off-by: Andrew Morton Signed-off-by: Linus Torvalds --- fs/aio.c | 47 ++++++++++++++++++++++++++++++++--------------- 1 file changed, 32 insertions(+), 15 deletions(-) (limited to 'fs') diff --git a/fs/aio.c b/fs/aio.c index 3428e9ae2f1d..d3bff60b5fe6 100644 --- a/fs/aio.c +++ b/fs/aio.c @@ -71,12 +71,6 @@ struct aio_ring_info { struct page *internal_pages[AIO_RING_PAGES]; }; -static inline unsigned aio_ring_avail(struct aio_ring_info *info, - struct aio_ring *ring) -{ - return (ring->head + info->nr - 1 - ring->tail) % info->nr; -} - struct kioctx { atomic_t users; atomic_t dead; @@ -92,7 +86,13 @@ struct kioctx { atomic_t reqs_active; struct list_head active_reqs; /* used for cancellation */ - /* sys_io_setup currently limits this to an unsigned int */ + /* + * This is what userspace passed to io_setup(), it's not used for + * anything but counting against the global max_reqs quota. + * + * The real limit is ring->nr - 1, which will be larger (see + * aio_setup_ring()) + */ unsigned max_reqs; struct aio_ring_info ring_info; @@ -284,8 +284,11 @@ static void free_ioctx_rcu(struct rcu_head *head) */ static void free_ioctx(struct kioctx *ctx) { + struct aio_ring_info *info = &ctx->ring_info; + struct aio_ring *ring; struct io_event res; struct kiocb *req; + unsigned head, avail; spin_lock_irq(&ctx->ctx_lock); @@ -299,7 +302,21 @@ static void free_ioctx(struct kioctx *ctx) spin_unlock_irq(&ctx->ctx_lock); - wait_event(ctx->wait, !atomic_read(&ctx->reqs_active)); + ring = kmap_atomic(info->ring_pages[0]); + head = ring->head; + kunmap_atomic(ring); + + while (atomic_read(&ctx->reqs_active) > 0) { + wait_event(ctx->wait, head != info->tail); + + avail = (head <= info->tail ? info->tail : info->nr) - head; + + atomic_sub(avail, &ctx->reqs_active); + head += avail; + head %= info->nr; + } + + WARN_ON(atomic_read(&ctx->reqs_active) < 0); aio_free_ring(ctx); @@ -548,7 +565,6 @@ static int kiocb_batch_refill(struct kioctx *ctx, struct kiocb_batch *batch) unsigned short allocated, to_alloc; long avail; struct kiocb *req, *n; - struct aio_ring *ring; to_alloc = min(batch->count, KIOCB_BATCH_SIZE); for (allocated = 0; allocated < to_alloc; allocated++) { @@ -563,10 +579,8 @@ static int kiocb_batch_refill(struct kioctx *ctx, struct kiocb_batch *batch) goto out; spin_lock_irq(&ctx->ctx_lock); - ring = kmap_atomic(ctx->ring_info.ring_pages[0]); - avail = aio_ring_avail(&ctx->ring_info, ring) - - atomic_read(&ctx->reqs_active); + avail = ctx->ring_info.nr - atomic_read(&ctx->reqs_active) - 1; BUG_ON(avail < 0); if (avail < allocated) { /* Trim back the number of requests. */ @@ -581,7 +595,6 @@ static int kiocb_batch_refill(struct kioctx *ctx, struct kiocb_batch *batch) batch->count -= allocated; atomic_add(allocated, &ctx->reqs_active); - kunmap_atomic(ring); spin_unlock_irq(&ctx->ctx_lock); out: @@ -688,8 +701,11 @@ void aio_complete(struct kiocb *iocb, long res, long res2) * when the event got cancelled. */ if (unlikely(xchg(&iocb->ki_cancel, - KIOCB_CANCELLED) == KIOCB_CANCELLED)) + KIOCB_CANCELLED) == KIOCB_CANCELLED)) { + atomic_dec(&ctx->reqs_active); + /* Still need the wake_up in case free_ioctx is waiting */ goto put_rq; + } /* * Add a completion event to the ring buffer. Must be done holding @@ -746,7 +762,6 @@ void aio_complete(struct kiocb *iocb, long res, long res2) put_rq: /* everything turned out well, dispose of the aiocb. */ aio_put_req(iocb); - atomic_dec(&ctx->reqs_active); /* * We have to order our ring_info tail store above and test @@ -825,6 +840,8 @@ static long aio_read_events_ring(struct kioctx *ctx, flush_dcache_page(info->ring_pages[0]); pr_debug("%li h%u t%u\n", ret, head, info->tail); + + atomic_sub(ret, &ctx->reqs_active); out: mutex_unlock(&info->ring_lock); -- cgit v1.2.3