Races in aio call back path.

@@ -609,7 +609,7 @@ static void unuse_mm(struct mm_struct *m
  * Should be called with the spin lock iocb->ki_ctx->ctx_lock
  * held
  */
-static inline int __queue_kicked_iocb(struct kiocb *iocb)
+static inline int queue_kicked_iocb(struct kiocb *iocb)
 {
 	struct kioctx *ctx = iocb->ki_ctx;
 

@@ -724,13 +724,6 @@ static ssize_t aio_run_iocb(struct kiocb
 			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);
 	}
 out:
 	spin_lock_irq(&ctx->ctx_lock);

@@ -741,17 +734,23 @@ out:
 		 * and know that there is more left to go,
 		 * this is where we let go so that a subsequent
 		 * "kick" can start the next iteration
+		 *
+		 * Issue an additional retry to avoid waiting forever if
+		 * no waits were queued (e.g. in case of a short read).
+		 * (Should be done with ctx_lock held.)
 		 */
 
-		/* will make __queue_kicked_iocb succeed from here on */
+		if (list_empty(&iocb->ki_wait.task_list))
+			kiocbSetKicked(iocb);
+		/* will make queue_kicked_iocb succeed from here on */
 		INIT_LIST_HEAD(&iocb->ki_run_list);
 		/* we must queue the next iteration ourselves, if it
 		 * has already been kicked */
 		if (kiocbIsKicked(iocb)) {
-			__queue_kicked_iocb(iocb);
+			queue_kicked_iocb(iocb);
 
 			/*
-			 * __queue_kicked_iocb will always return 1 here, because
+			 * queue_kicked_iocb will always return 1 here, because
 			 * iocb->ki_run_list is empty at this point so it should
 			 * be safe to unconditionally queue the context into the
 			 * work queue.

@@ -870,21 +869,31 @@ static void aio_kick_handler(void *data)
 
 
 /*
- * Called by kick_iocb to queue the kiocb for retry
- * and if required activate the aio work queue to process
- * it
+ * Kicking an async iocb.
+ * The following operations for the async iocbs should be atomic
+ * to avoid races with the aio_run_iocb() code.
+ * (1) Deleting the wait queue entry.
+ * (2) Kicking the iocb.
+ * (3) Queue the iocb back to run_list.
+ * Holds the ctx->ctx_lock to avoid races.
  */
-static void queue_kicked_iocb(struct kiocb *iocb)
+static void kick_async_iocb(struct kiocb *iocb)
 {
  	struct kioctx	*ctx = iocb->ki_ctx;
 	unsigned long flags;
 	int run = 0;
-
-	WARN_ON((!list_empty(&iocb->ki_wait.task_list)));
-
+	
 	spin_lock_irqsave(&ctx->ctx_lock, flags);
-	run = __queue_kicked_iocb(iocb);
+	list_del_init(&iocb->ki_wait.task_list);
+	/* If its already kicked we shouldn't queue it again */
+	if (!kiocbTryKick(iocb)) {
+		run = queue_kicked_iocb(iocb);
+	}
 	spin_unlock_irqrestore(&ctx->ctx_lock, flags);
+
+	/* Activate the aio worker queue if we have successfully queued
+	 * the iocb, so that it can be processed
+	 */
 	if (run)
 		aio_queue_work(ctx);
 }

@@ -901,15 +910,13 @@ void fastcall kick_iocb(struct kiocb *io
 	/* sync iocbs are easy: they can only ever be executing from a 
 	 * single context. */
 	if (is_sync_kiocb(iocb)) {
+		list_del_init(&iocb->ki_wait.task_list);
 		kiocbSetKicked(iocb);
 	        wake_up_process(iocb->ki_obj.tsk);
 		return;
-	}
-
-	/* If its already kicked we shouldn't queue it again */
-	if (!kiocbTryKick(iocb)) {
-		queue_kicked_iocb(iocb);
-	}
+	} else 
+		kick_async_iocb(iocb);
+	
 }
 EXPORT_SYMBOL(kick_iocb);
 

@@ -1461,7 +1468,6 @@ static int aio_wake_function(wait_queue_
 {
 	struct kiocb *iocb = container_of(wait, struct kiocb, ki_wait);
 
-	list_del_init(&wait->task_list);
 	kick_iocb(iocb);
 	return 1;
 }