[PATCH] aio: remove unlocked task_list test and resulting race

Only one of the run or kick path is supposed to put an iocb on the run
list.  If both of them do it than one of them can end up referencing a
freed iocb.  The kick path could delete the task_list item from the wait
queue before getting the ctx_lock and putting the iocb on the run list.
The run path was testing the task_list item outside the lock so that it
could catch ki_retry methods that return -EIOCBRETRY *without* putting the
iocb on a wait queue and promising to call kick_iocb.  This unlocked check
could then race with the kick path to cause both to try and put the iocb on
the run list.

The patch stops the run path from testing task_list by requring that any
ki_retry that returns -EIOCBRETRY *must* guarantee that kick_iocb() will be
called in the future.  aio_p{read,write}, the only in-tree -EIOCBRETRY
users, are updated.

Signed-off-by: Zach Brown <zach.brown@oracle.com>
Signed-off-by: Benjamin LaHaise <bcrl@linux.intel.com>
Signed-off-by: Andrew Morton <akpm@osdl.org>
Signed-off-by: Linus Torvalds <torvalds@osdl.org>

2 files changed, 67 insertions, 46 deletions

diff --git a/fs/aio.c b/fs/aio.c
index b8f296999c04..9edc0e4a1219 100644
--- a/fs/aio.c
+++ b/fs/aio.c
@@ -741,19 +741,9 @@ static ssize_t aio_run_iocb(struct kiocb *iocb)
         ret = retry(iocb);
         current->io_wait = NULL;
 
-        if (-EIOCBRETRY != ret) {
-                if (-EIOCBQUEUED != ret) {
-                        BUG_ON(!list_empty(&iocb->ki_wait.task_list));
-                        aio_complete(iocb, ret, 0);
-                        /* must not access the iocb after this */
-                }
-        } else {
-                /*
-                 * Issue an additional retry to avoid waiting forever if
-                 * no waits were queued (e.g. in case of a short read).
-                 */
-                if (list_empty(&iocb->ki_wait.task_list))
-                        kiocbSetKicked(iocb);
+        if (ret != -EIOCBRETRY && ret != -EIOCBQUEUED) {
+                BUG_ON(!list_empty(&iocb->ki_wait.task_list));
+                aio_complete(iocb, ret, 0);
         }
 out:
         spin_lock_irq(&ctx->ctx_lock);
@@ -1327,8 +1317,11 @@ asmlinkage long sys_io_destroy(aio_context_t ctx)
 }
 
 /*
- * Default retry method for aio_read (also used for first time submit)
- * Responsible for updating iocb state as retries progress
+ * aio_p{read,write} are the default  ki_retry methods for
+ * IO_CMD_P{READ,WRITE}.  They maintains kiocb retry state around potentially
+ * multiple calls to f_op->aio_read().  They loop around partial progress
+ * instead of returning -EIOCBRETRY because they don't have the means to call
+ * kick_iocb().
  */
 static ssize_t aio_pread(struct kiocb *iocb)
 {
@@ -1337,25 +1330,25 @@ static ssize_t aio_pread(struct kiocb *iocb)
         struct inode *inode = mapping->host;
         ssize_t ret = 0;
 
-        ret = file->f_op->aio_read(iocb, iocb->ki_buf,
-                iocb->ki_left, iocb->ki_pos);
+        do {
+                ret = file->f_op->aio_read(iocb, iocb->ki_buf,
+                        iocb->ki_left, iocb->ki_pos);
+                /*
+                 * Can't just depend on iocb->ki_left to determine
+                 * whether we are done. This may have been a short read.
+                 */
+                if (ret > 0) {
+                        iocb->ki_buf += ret;
+                        iocb->ki_left -= ret;
+                }
 
-        /*
-         * Can't just depend on iocb->ki_left to determine
-         * whether we are done. This may have been a short read.
-         */
-        if (ret > 0) {
-                iocb->ki_buf += ret;
-                iocb->ki_left -= ret;
                 /*
-                 * For pipes and sockets we return once we have
-                 * some data; for regular files we retry till we
-                 * complete the entire read or find that we can't
-                 * read any more data (e.g short reads).
+                 * For pipes and sockets we return once we have some data; for
+                 * regular files we retry till we complete the entire read or
+                 * find that we can't read any more data (e.g short reads).
                  */
-                if (!S_ISFIFO(inode->i_mode) && !S_ISSOCK(inode->i_mode))
-                        ret = -EIOCBRETRY;
-        }
+        } while (ret > 0 &&
+                 !S_ISFIFO(inode->i_mode) && !S_ISSOCK(inode->i_mode));
 
         /* This means we must have transferred all that we could */
         /* No need to retry anymore */
@@ -1365,27 +1358,21 @@ static ssize_t aio_pread(struct kiocb *iocb)
         return ret;
 }
 
-/*
- * Default retry method for aio_write (also used for first time submit)
- * Responsible for updating iocb state as retries progress
- */
+/* see aio_pread() */
 static ssize_t aio_pwrite(struct kiocb *iocb)
 {
         struct file *file = iocb->ki_filp;
         ssize_t ret = 0;
 
-        ret = file->f_op->aio_write(iocb, iocb->ki_buf,
-                iocb->ki_left, iocb->ki_pos);
-
-        if (ret > 0) {
-                iocb->ki_buf += ret;
-                iocb->ki_left -= ret;
-
-                ret = -EIOCBRETRY;
-        }
+        do {
+                ret = file->f_op->aio_write(iocb, iocb->ki_buf,
+                        iocb->ki_left, iocb->ki_pos);
+                if (ret > 0) {
+                        iocb->ki_buf += ret;
+                        iocb->ki_left -= ret;
+                }
+        } while (ret > 0);
 
-        /* This means we must have transferred all that we could */
-        /* No need to retry anymore */
         if ((ret == 0) || (iocb->ki_left == 0))
                 ret = iocb->ki_nbytes - iocb->ki_left;
 
diff --git a/include/linux/aio.h b/include/linux/aio.h
index a4d5af907f90..60def658b246 100644
--- a/include/linux/aio.h
+++ b/include/linux/aio.h
@@ -43,6 +43,40 @@ struct kioctx;
 #define kiocbIsKicked(iocb)     test_bit(KIF_KICKED, &(iocb)->ki_flags)
 #define kiocbIsCancelled(iocb)  test_bit(KIF_CANCELLED, &(iocb)->ki_flags)
 
+/* is there a better place to document function pointer methods? */
+/**
+ * ki_retry     -       iocb forward progress callback
+ * @kiocb:      The kiocb struct to advance by performing an operation.
+ *
+ * This callback is called when the AIO core wants a given AIO operation
+ * to make forward progress.  The kiocb argument describes the operation
+ * that is to be performed.  As the operation proceeds, perhaps partially,
+ * ki_retry is expected to update the kiocb with progress made.  Typically
+ * ki_retry is set in the AIO core and it itself calls file_operations
+ * helpers.
+ *
+ * ki_retry's return value determines when the AIO operation is completed
+ * and an event is generated in the AIO event ring.  Except the special
+ * return values described below, the value that is returned from ki_retry
+ * is transferred directly into the completion ring as the operation's
+ * resulting status.  Once this has happened ki_retry *MUST NOT* reference
+ * the kiocb pointer again.
+ *
+ * If ki_retry returns -EIOCBQUEUED it has made a promise that aio_complete()
+ * will be called on the kiocb pointer in the future.  The AIO core will
+ * not ask the method again -- ki_retry must ensure forward progress.
+ * aio_complete() must be called once and only once in the future, multiple
+ * calls may result in undefined behaviour.
+ *
+ * If ki_retry returns -EIOCBRETRY it has made a promise that kick_iocb()
+ * will be called on the kiocb pointer in the future.  This may happen
+ * through generic helpers that associate kiocb->ki_wait with a wait
+ * queue head that ki_retry uses via current->io_wait.  It can also happen
+ * with custom tracking and manual calls to kick_iocb(), though that is
+ * discouraged.  In either case, kick_iocb() must be called once and only
+ * once.  ki_retry must ensure forward progress, the AIO core will wait
+ * indefinitely for kick_iocb() to be called.
+ */
 struct kiocb {
         struct list_head        ki_run_list;
         long                    ki_flags;