1 files changed, 85 insertions, 5 deletions

diff --git a/fs/aio.c b/fs/aio.c
index ae635872affb..733750096b71 100644
--- a/fs/aio.c
+++ b/fs/aio.c
@@ -141,6 +141,7 @@ struct kioctx {
 
         struct {
                 unsigned        tail;
+                unsigned        completed_events;
                 spinlock_t      completion_lock;
         } ____cacheline_aligned_in_smp;
 
@@ -792,6 +793,8 @@ void exit_aio(struct mm_struct *mm)
 
         for (i = 0; i < table->nr; ++i) {
                 struct kioctx *ctx = table->table[i];
+                struct completion requests_done =
+                        COMPLETION_INITIALIZER_ONSTACK(requests_done);
 
                 if (!ctx)
                         continue;
@@ -803,7 +806,10 @@ void exit_aio(struct mm_struct *mm)
                  * that it needs to unmap the area, just set it to 0.
                  */
                 ctx->mmap_size = 0;
-                kill_ioctx(mm, ctx, NULL);
+                kill_ioctx(mm, ctx, &requests_done);
+
+                /* Wait until all IO for the context are done. */
+                wait_for_completion(&requests_done);
         }
 
         RCU_INIT_POINTER(mm->ioctx_table, NULL);
@@ -857,6 +863,68 @@ out:
         return ret;
 }
 
+/* refill_reqs_available
+ *      Updates the reqs_available reference counts used for tracking the
+ *      number of free slots in the completion ring.  This can be called
+ *      from aio_complete() (to optimistically update reqs_available) or
+ *      from aio_get_req() (the we're out of events case).  It must be
+ *      called holding ctx->completion_lock.
+ */
+static void refill_reqs_available(struct kioctx *ctx, unsigned head,
+                                  unsigned tail)
+{
+        unsigned events_in_ring, completed;
+
+        /* Clamp head since userland can write to it. */
+        head %= ctx->nr_events;
+        if (head <= tail)
+                events_in_ring = tail - head;
+        else
+                events_in_ring = ctx->nr_events - (head - tail);
+
+        completed = ctx->completed_events;
+        if (events_in_ring < completed)
+                completed -= events_in_ring;
+        else
+                completed = 0;
+
+        if (!completed)
+                return;
+
+        ctx->completed_events -= completed;
+        put_reqs_available(ctx, completed);
+}
+
+/* user_refill_reqs_available
+ *      Called to refill reqs_available when aio_get_req() encounters an
+ *      out of space in the completion ring.
+ */
+static void user_refill_reqs_available(struct kioctx *ctx)
+{
+        spin_lock_irq(&ctx->completion_lock);
+        if (ctx->completed_events) {
+                struct aio_ring *ring;
+                unsigned head;
+
+                /* Access of ring->head may race with aio_read_events_ring()
+                 * here, but that's okay since whether we read the old version
+                 * or the new version, and either will be valid.  The important
+                 * part is that head cannot pass tail since we prevent
+                 * aio_complete() from updating tail by holding
+                 * ctx->completion_lock.  Even if head is invalid, the check
+                 * against ctx->completed_events below will make sure we do the
+                 * safe/right thing.
+                 */
+                ring = kmap_atomic(ctx->ring_pages[0]);
+                head = ring->head;
+                kunmap_atomic(ring);
+
+                refill_reqs_available(ctx, head, ctx->tail);
+        }
+
+        spin_unlock_irq(&ctx->completion_lock);
+}
+
 /* aio_get_req
  *      Allocate a slot for an aio request.
  * Returns NULL if no requests are free.
@@ -865,8 +933,11 @@ static inline struct kiocb *aio_get_req(struct kioctx *ctx)
 {
         struct kiocb *req;
 
-        if (!get_reqs_available(ctx))
-                return NULL;
+        if (!get_reqs_available(ctx)) {
+                user_refill_reqs_available(ctx);
+                if (!get_reqs_available(ctx))
+                        return NULL;
+        }
 
         req = kmem_cache_alloc(kiocb_cachep, GFP_KERNEL|__GFP_ZERO);
         if (unlikely(!req))
@@ -925,8 +996,8 @@ void aio_complete(struct kiocb *iocb, long res, long res2)
         struct kioctx   *ctx = iocb->ki_ctx;
         struct aio_ring *ring;
         struct io_event *ev_page, *event;
+        unsigned tail, pos, head;
         unsigned long   flags;
-        unsigned tail, pos;
 
         /*
          * Special case handling for sync iocbs:
@@ -987,10 +1058,14 @@ void aio_complete(struct kiocb *iocb, long res, long res2)
         ctx->tail = tail;
 
         ring = kmap_atomic(ctx->ring_pages[0]);
+        head = ring->head;
         ring->tail = tail;
         kunmap_atomic(ring);
         flush_dcache_page(ctx->ring_pages[0]);
 
+        ctx->completed_events++;
+        if (ctx->completed_events > 1)
+                refill_reqs_available(ctx, head, tail);
         spin_unlock_irqrestore(&ctx->completion_lock, flags);
 
         pr_debug("added to ring %p at [%u]\n", iocb, tail);
@@ -1005,7 +1080,6 @@ void aio_complete(struct kiocb *iocb, long res, long res2)
 
         /* everything turned out well, dispose of the aiocb. */
         kiocb_free(iocb);
-        put_reqs_available(ctx, 1);
 
         /*
          * We have to order our ring_info tail store above and test
@@ -1042,6 +1116,12 @@ static long aio_read_events_ring(struct kioctx *ctx,
         tail = ring->tail;
         kunmap_atomic(ring);
 
+        /*
+         * Ensure that once we've read the current tail pointer, that
+         * we also see the events that were stored up to the tail.
+         */
+        smp_rmb();
+
         pr_debug("h%u t%u m%u\n", head, tail, ctx->nr_events);
 
         if (head == tail)