[PATCH -mm 0/10][RFC] aio: make struct kiocb private

@@ -131,10 +131,32 @@ blkdev_get_block(struct inode *inode, se
 	return 0;
 }
 
-static int blk_end_aio(struct bio *bio, unsigned int bytes_done, int error)
+struct bdev_aio {
+	atomic_t		iocount;	/* refcount */
+	atomic_t		bytes_done;	/* byte counter */
+	int			err;		/* error handling */
+	file_endio_t		*endio;		/* end I/O notify fn */
+	void			*endio_data;	/* notify fn private data */
+};
+
+static void blk_io_put(struct bdev_aio *io)
+{
+	if (!atomic_dec_and_test(&io->iocount))
+		return;
+
+	if (!io->endio)
+		return complete((struct completion*)io->endio_data);
+
+	io->endio(io->endio_data, atomic_read(&io->bytes_done), io->err);
+	kfree(io);
+}
+
+static int blk_bio_endio(struct bio *bio, unsigned int bytes_done, int error)
 {
-	struct kiocb *iocb = bio->bi_private;
-	atomic_t *bio_count = &iocb->ki_bio_count;
+	struct bdev_aio *io = bio->bi_private;
+
+	if (bio->bi_size)
+		return 1;
 
 	if (bio_data_dir(bio) == READ)
 		bio_check_pages_dirty(bio);

@@ -143,16 +165,21 @@ static int blk_end_aio(struct bio *bio, 
 		bio_put(bio);
 	}
 
-	/* iocb->ki_nbytes stores error code from LLDD */
-	if (error)
-		iocb->ki_nbytes = -EIO;
-
-	if (atomic_dec_and_test(bio_count))
-		aio_complete(iocb, iocb->ki_left, iocb->ki_nbytes);
+  	if (error)
+		io->err = error;
+	atomic_add(bytes_done, &io->bytes_done);
 
+	blk_io_put(io);
 	return 0;
 }
 
+static void blk_io_init(struct bdev_aio *io)
+{
+	atomic_set(&io->iocount, 1);
+	atomic_set(&io->bytes_done, 0);
+	io->err = 0;
+}
+
 #define VEC_SIZE	16
 struct pvec {
 	unsigned short nr;

@@ -208,24 +235,33 @@ blkdev_direct_IO(int rw, struct kiocb *i
 
 	unsigned long addr;	/* user iovec address */
 	size_t count;		/* user iovec len */
-	size_t nbytes = iocb->ki_nbytes = iocb->ki_left; /* total xfer size */
+	size_t nbytes;		 /* total xfer size */
 	loff_t size;		/* size of block device */
 	struct bio *bio;
-	atomic_t *bio_count = &iocb->ki_bio_count;
+	struct bdev_aio stack_io, *io;
+	file_endio_t *endio = aio_complete;
+	void *endio_data = iocb;
 	struct page *page;
 	struct pvec pvec;
 
 	pvec.nr = 0;
 	pvec.idx = 0;
 
+	io = &stack_io;
+	if (endio) {
+		io = kmalloc(sizeof(struct bdev_aio), GFP_KERNEL);
+		if (!io)
+			return -ENOMEM;
+	}
+	blk_io_init(io);
+
 	if (pos & blocksize_mask)
 		return -EINVAL;
 
+	nbytes = iov_length(iov, nr_segs);
 	size = i_size_read(inode);
-	if (pos + nbytes > size) {
+	if (pos + nbytes > size)
 		nbytes = size - pos;
-		iocb->ki_left = nbytes;
-	}
 
 	/*
 	 * check first non-zero iov alignment, the remaining

@@ -237,7 +273,6 @@ blkdev_direct_IO(int rw, struct kiocb *i
 		if (addr & blocksize_mask || count & blocksize_mask)
 			return -EINVAL;
 	} while (!count && ++seg < nr_segs);
-	atomic_set(bio_count, 1);
 
 	while (nbytes) {
 		/* roughly estimate number of bio vec needed */

@@ -248,8 +283,8 @@ blkdev_direct_IO(int rw, struct kiocb *i
 		/* bio_alloc should not fail with GFP_KERNEL flag */
 		bio = bio_alloc(GFP_KERNEL, nvec);
 		bio->bi_bdev = I_BDEV(inode);
-		bio->bi_end_io = blk_end_aio;
-		bio->bi_private = iocb;
+		bio->bi_end_io = blk_bio_endio;
+		bio->bi_private = io;
 		bio->bi_sector = pos >> blkbits;
 same_bio:
 		cur_off = addr & ~PAGE_MASK;

@@ -289,18 +324,27 @@ same_bio:
 		/* bio is ready, submit it */
 		if (rw == READ)
 			bio_set_pages_dirty(bio);
-		atomic_inc(bio_count);
+		atomic_inc(&io->iocount);
 		submit_bio(rw, bio);
 	}
 
 completion:
-	iocb->ki_left -= nbytes;
-	nbytes = iocb->ki_left;
-	iocb->ki_pos += nbytes;
+	if (!endio) {
+		struct completion event;
+
+		init_completion(&event);
+		io->endio = NULL;
+		io->endio_data = &event;
+
+		if (!atomic_dec_and_test(&io->iocount))
+			wait_for_completion(&event);
+		return io->err ? io->err : atomic_read(&io->bytes_done);
+	}
 
-	if (atomic_dec_and_test(bio_count))
-		aio_complete(iocb, nbytes, 0);
+	io->endio = endio;
+	io->endio_data = endio_data;
 
+	blk_io_put(io);
 	return -EIOCBQUEUED;
 
 backout:

@@ -316,7 +360,7 @@ backout:
 	 * if no bio was submmitted, return the error code.
 	 * otherwise, proceed with pending I/O completion.
 	 */
-	if (atomic_read(bio_count) == 1)
+	if (atomic_read(&io->iocount) == 1)
 		return PTR_ERR(page);
 	goto completion;
 }

--

@@ -559,35 +559,32 @@ static int ep_aio_cancel(struct kiocb *i
 	return value;
 }
 
-static ssize_t ep_aio_read_retry(struct kiocb *iocb)
+static int ep_aio_read_retry(struct kiocb *iocb)
 {
 	struct kiocb_priv	*priv = iocb->private;
-	ssize_t			len, total;
-	int			i;
+	ssize_t			total;
+	int			i, err = 0;
 
   	/* we "retry" to get the right mm context for this: */
 
  	/* copy stuff into user buffers */
  	total = priv->actual;
- 	len = 0;
  	for (i=0; i < priv->nr_segs; i++) {
  		ssize_t this = min((ssize_t)(priv->iv[i].iov_len), total);
 
  		if (copy_to_user(priv->iv[i].iov_base, priv->buf, this)) {
- 			if (len == 0)
- 				len = -EFAULT;
+ 			err = -EFAULT;
  			break;
  		}
 
  		total -= this;
- 		len += this;
  		if (total == 0)
  			break;
  	}
   	kfree(priv->buf);
   	kfree(priv);
   	aio_put_req(iocb);
- 	return len;
+ 	return err;
 }
 
 static void ep_aio_complete(struct usb_ep *ep, struct usb_request *req)

@@ -610,9 +607,7 @@ static void ep_aio_complete(struct usb_e
 		if (unlikely(kiocbIsCancelled(iocb)))
 			aio_put_req(iocb);
 		else
-			aio_complete(iocb,
-				req->actual ? req->actual : req->status,
-				req->status);
+			aio_complete(iocb, req->actual, req->status);
 	} else {
 		/* retry() won't report both; so we hide some faults */
 		if (unlikely(0 != req->status))

@@ -702,16 +697,17 @@ ep_aio_read(struct kiocb *iocb, const st
 {
 	struct ep_data		*epdata = iocb->ki_filp->private_data;
 	char			*buf;
+	size_t			len = iov_length(iov, nr_segs);
 
 	if (unlikely(epdata->desc.bEndpointAddress & USB_DIR_IN))
 		return -EINVAL;
 
-	buf = kmalloc(iocb->ki_left, GFP_KERNEL);
+	buf = kmalloc(len, GFP_KERNEL);
 	if (unlikely(!buf))
 		return -ENOMEM;
 
 	iocb->ki_retry = ep_aio_read_retry;
-	return ep_aio_rwtail(iocb, buf, iocb->ki_left, epdata, iov, nr_segs);
+	return ep_aio_rwtail(iocb, buf, len, epdata, iov, nr_segs);
 }
 
 static ssize_t

@@ -726,7 +722,7 @@ ep_aio_write(struct kiocb *iocb, const s
 	if (unlikely(!(epdata->desc.bEndpointAddress & USB_DIR_IN)))
 		return -EINVAL;
 
-	buf = kmalloc(iocb->ki_left, GFP_KERNEL);
+	buf = kmalloc(iov_length(iov, nr_segs), GFP_KERNEL);
 	if (unlikely(!buf))
 		return -ENOMEM;
 

@@ -658,16 +658,16 @@ static inline int __queue_kicked_iocb(st
  * simplifies the coding of individual aio operations as
  * it avoids various potential races.
  */
-static ssize_t aio_run_iocb(struct kiocb *iocb)
+static void aio_run_iocb(struct kiocb *iocb)
 {
 	struct kioctx	*ctx = iocb->ki_ctx;
-	ssize_t (*retry)(struct kiocb *);
+	int (*retry)(struct kiocb *);
 	wait_queue_t *io_wait = current->io_wait;
-	ssize_t ret;
+	int err;
 
 	if (!(retry = iocb->ki_retry)) {
 		printk("aio_run_iocb: iocb->ki_retry = NULL\n");
-		return 0;
+		return;
 	}
 
 	/*

@@ -702,8 +702,8 @@ static ssize_t aio_run_iocb(struct kiocb
 
 	/* Quit retrying if the i/o has been cancelled */
 	if (kiocbIsCancelled(iocb)) {
-		ret = -EINTR;
-		aio_complete(iocb, ret, 0);
+		err = -EINTR;
+		aio_complete(iocb, iocb->ki_nbytes - iocb->ki_left, err);
 		/* must not access the iocb after this */
 		goto out;
 	}

@@ -720,17 +720,17 @@ static ssize_t aio_run_iocb(struct kiocb
 	 */
 	BUG_ON(!is_sync_wait(current->io_wait));
 	current->io_wait = &iocb->ki_wait.wait;
-	ret = retry(iocb);
+	err = retry(iocb);
 	current->io_wait = io_wait;
 
-	if (ret != -EIOCBRETRY && ret != -EIOCBQUEUED) {
+	if (err != -EIOCBRETRY && err != -EIOCBQUEUED) {
 		BUG_ON(!list_empty(&iocb->ki_wait.wait.task_list));
-		aio_complete(iocb, ret, 0);
+		aio_complete(iocb, iocb->ki_nbytes - iocb->ki_left, err);
 	}
 out:
 	spin_lock_irq(&ctx->ctx_lock);
 
-	if (-EIOCBRETRY == ret) {
+	if (-EIOCBRETRY == err) {
 		/*
 		 * OK, now that we are done with this iteration
 		 * and know that there is more left to go,

@@ -754,7 +754,6 @@ out:
 			aio_queue_work(ctx);
 		}
 	}
-	return ret;
 }
 
 /*

@@ -918,19 +917,25 @@ EXPORT_SYMBOL(kick_iocb);
 
 /* aio_complete
  *	Called when the io request on the given iocb is complete.
- *	Returns true if this is the last user of the request.  The 
+ *	Frees ioctx if this is the last user of the request.  The 
  *	only other user of the request can be the cancellation code.
  */
-int fastcall aio_complete(struct kiocb *iocb, long res, long res2)
+void fastcall aio_complete(void *endio_data, ssize_t count, int err)
 {
+	struct kiocb *iocb = endio_data;
 	struct kioctx	*ctx = iocb->ki_ctx;
 	struct aio_ring_info	*info;
 	struct aio_ring	*ring;
 	struct io_event	*event;
 	unsigned long	flags;
 	unsigned long	tail;
+	long		result;
 	int		ret;
 
+	result = (long) err;
+	if (!result)
+		result = (long) count;
+
 	/*
 	 * Special case handling for sync iocbs:
 	 *  - events go directly into the iocb for fast handling

@@ -940,10 +945,10 @@ int fastcall aio_complete(struct kiocb *
 	 */
 	if (is_sync_kiocb(iocb)) {
 		BUG_ON(iocb->ki_users != 1);
-		iocb->ki_user_data = res;
+		iocb->ki_user_data = result;
 		iocb->ki_users = 0;
 		wake_up_process(iocb->ki_obj.tsk);
-		return 1;
+		return;
 	}
 
 	info = &ctx->ring_info;

@@ -975,12 +980,12 @@ int fastcall aio_complete(struct kiocb *
 
 	event->obj = (u64)(unsigned long)iocb->ki_obj.user;
 	event->data = iocb->ki_user_data;
-	event->res = res;
-	event->res2 = res2;
+	event->res = result;
+	event->res2 = err;
 
 	dprintk("aio_complete: %p[%lu]: %p: %p %Lx %lx %lx\n",
 		ctx, tail, iocb, iocb->ki_obj.user, iocb->ki_user_data,
-		res, res2);
+		result, err);
 
 	/* after flagging the request as done, we
 	 * must never even look at it again

@@ -1002,7 +1007,7 @@ put_rq:
 		wake_up(&ctx->wait);
 
 	spin_unlock_irqrestore(&ctx->ctx_lock, flags);
-	return ret;
+	return;
 }
 
 /* aio_read_evt

@@ -1307,7 +1312,7 @@ static void aio_advance_iovec(struct kio
 	BUG_ON(ret > 0 && iocb->ki_left == 0);
 }
 
-static ssize_t aio_rw_vect_retry(struct kiocb *iocb)
+static int aio_rw_vect_retry(struct kiocb *iocb)
 {
 	struct file *file = iocb->ki_filp;
 	struct address_space *mapping = file->f_mapping;

@@ -1341,26 +1346,26 @@ static ssize_t aio_rw_vect_retry(struct 
 
 	/* 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;
+	if (iocb->ki_left == 0)
+		ret = 0;
 
-	return ret;
+	return (int) ret;
 }
 
-static ssize_t aio_fdsync(struct kiocb *iocb)
+static int aio_fdsync(struct kiocb *iocb)
 {
 	struct file *file = iocb->ki_filp;
-	ssize_t ret = -EINVAL;
+	int ret = -EINVAL;
 
 	if (file->f_op->aio_fsync)
 		ret = file->f_op->aio_fsync(iocb, 1);
 	return ret;
 }
 
-static ssize_t aio_fsync(struct kiocb *iocb)
+static int aio_fsync(struct kiocb *iocb)
 {
 	struct file *file = iocb->ki_filp;
-	ssize_t ret = -EINVAL;
+	int ret = -EINVAL;
 
 	if (file->f_op->aio_fsync)
 		ret = file->f_op->aio_fsync(iocb, 0);

@@ -147,12 +147,8 @@ static int blk_end_aio(struct bio *bio, 
 	if (error)
 		iocb->ki_nbytes = -EIO;
 
-	if (atomic_dec_and_test(bio_count)) {
-		if ((long)iocb->ki_nbytes < 0)
-			aio_complete(iocb, iocb->ki_nbytes, 0);
-		else
-			aio_complete(iocb, iocb->ki_left, 0);
-	}
+	if (atomic_dec_and_test(bio_count))
+		aio_complete(iocb, iocb->ki_left, iocb->ki_nbytes);
 
 	return 0;
 }

@@ -224,8 +224,6 @@ static struct page *dio_get_page(struct 
  */
 static int dio_complete(struct dio *dio, loff_t offset, int ret)
 {
-	ssize_t transferred = 0;
-
 	/*
 	 * AIO submission can race with bio completion to get here while
 	 * expecting to have the last io completed by bio completion.

@@ -236,15 +234,13 @@ static int dio_complete(struct dio *dio,
 		ret = 0;
 
 	if (dio->result) {
-		transferred = dio->result;
-
 		/* Check for short read case */
-		if ((dio->rw == READ) && ((offset + transferred) > dio->i_size))
-			transferred = dio->i_size - offset;
+		if ((dio->rw == READ) && ((offset + dio->result) > dio->i_size))
+			dio->result = dio->i_size - offset;
 	}
 
 	if (dio->end_io && dio->result)
-		dio->end_io(dio->iocb, offset, transferred,
+		dio->end_io(dio->iocb, offset, dio->result,
 			    dio->map_bh.b_private);
 	if (dio->lock_type == DIO_LOCKING)
 		/* lockdep: non-owner release */

@@ -254,8 +250,6 @@ static int dio_complete(struct dio *dio,
 		ret = dio->page_errors;
 	if (ret == 0)
 		ret = dio->io_error;
-	if (ret == 0)
-		ret = transferred;
 
 	return ret;
 }

@@ -283,8 +277,8 @@ static int dio_bio_end_aio(struct bio *b
 	spin_unlock_irqrestore(&dio->bio_lock, flags);
 
 	if (remaining == 0) {
-		int ret = dio_complete(dio, dio->iocb->ki_pos, 0);
-		aio_complete(dio->iocb, ret, 0);
+		int err = dio_complete(dio, dio->iocb->ki_pos, 0);
+		aio_complete(dio->iocb, dio->result, err);
 		kfree(dio);
 	}
 

@@ -1110,6 +1104,8 @@ direct_io_worker(int rw, struct kiocb *i
 	BUG_ON(!dio->is_async && ret2 != 0);
 	if (ret2 == 0) {
 		ret = dio_complete(dio, offset, ret);
+		if (ret == 0)
+			ret = dio->result;	/* bytes transferred */
 		kfree(dio);
 	} else
 		BUG_ON(ret != -EIOCBQUEUED);

@@ -200,12 +200,9 @@ out:
  */
 static void nfs_direct_complete(struct nfs_direct_req *dreq)
 {
-	if (dreq->iocb) {
-		long res = (long) dreq->error;
-		if (!res)
-			res = (long) dreq->count;
-		aio_complete(dreq->iocb, res, 0);
-	}
+	if (dreq->iocb)
+		aio_complete(dreq->iocb, dreq->count, dreq->error);
+
 	complete_all(&dreq->completion);
 
 	kref_put(&dreq->kref, nfs_direct_req_release);

@@ -15,10 +15,9 @@
 struct kioctx;
 
 /* Notes on cancelling a kiocb:
- *	If a kiocb is cancelled, aio_complete may return 0 to indicate 
- *	that cancel has not yet disposed of the kiocb.  All cancel 
- *	operations *must* call aio_put_req to dispose of the kiocb 
- *	to guard against races with the completion code.
+ *	If a kiocb is cancelled, aio_complete may not yet have disposed of
+ *	the kiocb.  All cancel operations *must* call aio_put_req to dispose
+ *	of the kiocb to guard against races with the completion code.
  */
 #define KIOCB_C_CANCELLED	0x01
 #define KIOCB_C_COMPLETE	0x02

@@ -98,7 +97,7 @@ struct kiocb {
 	struct file		*ki_filp;
 	struct kioctx		*ki_ctx;	/* may be NULL for sync ops */
 	int			(*ki_cancel)(struct kiocb *, struct io_event *);
-	ssize_t			(*ki_retry)(struct kiocb *);
+	int			(*ki_retry)(struct kiocb *);
 	void			(*ki_dtor)(struct kiocb *);
 
 	union {

@@ -208,7 +207,7 @@ extern unsigned aio_max_size;
 extern ssize_t FASTCALL(wait_on_sync_kiocb(struct kiocb *iocb));
 extern int FASTCALL(aio_put_req(struct kiocb *iocb));
 extern void FASTCALL(kick_iocb(struct kiocb *iocb));
-extern int FASTCALL(aio_complete(struct kiocb *iocb, long res, long res2));
+extern void FASTCALL(aio_complete(void *endio_data, ssize_t count, int err));
 extern void FASTCALL(__put_ioctx(struct kioctx *ctx));
 struct mm_struct;
 extern void FASTCALL(exit_aio(struct mm_struct *mm));

@@ -1109,6 +1109,8 @@ typedef int (*read_actor_t)(read_descrip
 #define HAVE_COMPAT_IOCTL 1
 #define HAVE_UNLOCKED_IOCTL 1
 
+typedef void (file_endio_t)(void *endio_data, ssize_t count, int err);
+
 /*
  * NOTE:
  * read, write, poll, fsync, readv, writev, unlocked_ioctl and compat_ioctl

--

@@ -68,7 +68,6 @@ struct nfs_direct_req {
 
 	/* I/O parameters */
 	struct nfs_open_context	*ctx;		/* file open context info */
-	struct kiocb *		iocb;		/* controlling i/o request */
 	struct inode *		inode;		/* target file of i/o */
 
 	/* completion state */

@@ -77,6 +76,8 @@ struct nfs_direct_req {
 	ssize_t			count,		/* bytes actually processed */
 				error;		/* any reported error */
 	struct completion	completion;	/* wait for i/o completion */
+	file_endio_t		*endio;		/* async completion function */
+	void			*endio_data;	/* private completion data */
 
 	/* commit state */
 	struct list_head	rewrite_list;	/* saved nfs_write_data structs */

@@ -151,7 +152,7 @@ static inline struct nfs_direct_req *nfs
 	kref_get(&dreq->kref);
 	init_completion(&dreq->completion);
 	INIT_LIST_HEAD(&dreq->rewrite_list);
-	dreq->iocb = NULL;
+	dreq->endio = NULL;
 	dreq->ctx = NULL;
 	spin_lock_init(&dreq->lock);
 	atomic_set(&dreq->io_count, 0);

@@ -179,7 +180,7 @@ static ssize_t nfs_direct_wait(struct nf
 	ssize_t result = -EIOCBQUEUED;
 
 	/* Async requests don't wait here */
-	if (dreq->iocb)
+	if (!dreq->endio)
 		goto out;
 
 	result = wait_for_completion_interruptible(&dreq->completion);

@@ -194,14 +195,10 @@ out:
 	return (ssize_t) result;
 }
 
-/*
- * Synchronous I/O uses a stack-allocated iocb.  Thus we can't trust
- * the iocb is still valid here if this is a synchronous request.
- */
 static void nfs_direct_complete(struct nfs_direct_req *dreq)
 {
-	if (dreq->iocb)
-		aio_complete(dreq->iocb, dreq->count, dreq->error);
+	if (dreq->endio)
+		dreq->endio(dreq->endio_data, dreq->count, dreq->error);
 
 	complete_all(&dreq->completion);
 

@@ -332,11 +329,13 @@ static ssize_t nfs_direct_read_schedule(
 	return result < 0 ? (ssize_t) result : -EFAULT;
 }
 
-static ssize_t nfs_direct_read(struct kiocb *iocb, unsigned long user_addr, size_t count, loff_t pos)
+static ssize_t nfs_direct_read(struct file *file, unsigned long user_addr,
+			       size_t count, loff_t pos,
+			       file_endio_t *endio, void *endio_data)
 {
 	ssize_t result = 0;
 	sigset_t oldset;
-	struct inode *inode = iocb->ki_filp->f_mapping->host;
+	struct inode *inode = file->f_mapping->host;
 	struct rpc_clnt *clnt = NFS_CLIENT(inode);
 	struct nfs_direct_req *dreq;
 

@@ -345,9 +344,9 @@ static ssize_t nfs_direct_read(struct ki
 		return -ENOMEM;
 
 	dreq->inode = inode;
-	dreq->ctx = get_nfs_open_context((struct nfs_open_context *)iocb->ki_filp->private_data);
-	if (!is_sync_kiocb(iocb))
-		dreq->iocb = iocb;
+	dreq->ctx = get_nfs_open_context((struct nfs_open_context *)file->private_data);
+	dreq->endio = endio;
+	dreq->endio_data = endio_data;
 
 	nfs_add_stats(inode, NFSIOS_DIRECTREADBYTES, count);
 	rpc_clnt_sigmask(clnt, &oldset);

@@ -663,11 +662,13 @@ static ssize_t nfs_direct_write_schedule
 	return result < 0 ? (ssize_t) result : -EFAULT;
 }
 
-static ssize_t nfs_direct_write(struct kiocb *iocb, unsigned long user_addr, size_t count, loff_t pos)
+static ssize_t nfs_direct_write(struct file *file, unsigned long user_addr,
+				size_t count, loff_t pos,
+				file_endio_t *endio, void *endio_data)
 {
 	ssize_t result = 0;
 	sigset_t oldset;
-	struct inode *inode = iocb->ki_filp->f_mapping->host;
+	struct inode *inode = file->f_mapping->host;
 	struct rpc_clnt *clnt = NFS_CLIENT(inode);
 	struct nfs_direct_req *dreq;
 	size_t wsize = NFS_SERVER(inode)->wsize;

@@ -682,9 +683,9 @@ static ssize_t nfs_direct_write(struct k
 		sync = FLUSH_STABLE;
 
 	dreq->inode = inode;
-	dreq->ctx = get_nfs_open_context((struct nfs_open_context *)iocb->ki_filp->private_data);
-	if (!is_sync_kiocb(iocb))
-		dreq->iocb = iocb;
+	dreq->ctx = get_nfs_open_context((struct nfs_open_context *)file->private_data);
+	dreq->endio = endio;
+	dreq->endio_data = endio_data;
 
 	nfs_add_stats(inode, NFSIOS_DIRECTWRITTENBYTES, count);
 

@@ -701,10 +702,12 @@ static ssize_t nfs_direct_write(struct k
 
 /**
  * nfs_file_direct_read - file direct read operation for NFS files
- * @iocb: target I/O control block
+ * @file: target file
  * @iov: vector of user buffers into which to read data
  * @nr_segs: size of iov vector
  * @pos: byte offset in file where reading starts
+ * @endio: async I/O completion function
+ * @endio_data: private completion data
  *
  * We use this function for direct reads instead of calling
  * generic_file_aio_read() in order to avoid gfar's check to see if

@@ -720,11 +723,11 @@ static ssize_t nfs_direct_write(struct k
  * client must read the updated atime from the server back into its
  * cache.
  */
-ssize_t nfs_file_direct_read(struct kiocb *iocb, const struct iovec *iov,
-				unsigned long nr_segs, loff_t pos)
+ssize_t nfs_file_direct_read(struct file *file, const struct iovec *iov,
+				unsigned long nr_segs, loff_t *pos,
+				file_endio_t *endio, void *endio_data)
 {
 	ssize_t retval = -EINVAL;
-	struct file *file = iocb->ki_filp;
 	struct address_space *mapping = file->f_mapping;
 	/* XXX: temporary */
 	const char __user *buf = iov[0].iov_base;

@@ -733,7 +736,7 @@ ssize_t nfs_file_direct_read(struct kioc
 	dprintk("nfs: direct read(%s/%s, %lu@%Ld)\n",
 		file->f_path.dentry->d_parent->d_name.name,
 		file->f_path.dentry->d_name.name,
-		(unsigned long) count, (long long) pos);
+		(unsigned long) count, (long long) *pos);
 
 	if (nr_segs != 1)
 		return -EINVAL;

@@ -751,9 +754,10 @@ ssize_t nfs_file_direct_read(struct kioc
 	if (retval)
 		goto out;
 
-	retval = nfs_direct_read(iocb, (unsigned long) buf, count, pos);
+	retval = nfs_direct_read(file, (unsigned long) buf, count, *pos,
+				 endio, endio_data);
 	if (retval > 0)
-		iocb->ki_pos = pos + retval;
+		*pos += retval;
 
 out:
 	return retval;

@@ -761,10 +765,12 @@ out:
 
 /**
  * nfs_file_direct_write - file direct write operation for NFS files
- * @iocb: target I/O control block
+ * @file: target file
  * @iov: vector of user buffers from which to write data
  * @nr_segs: size of iov vector
  * @pos: byte offset in file where writing starts
+ * @endio: async I/O completion function
+ * @endio_data: private completion data
  *
  * We use this function for direct writes instead of calling
  * generic_file_aio_write() in order to avoid taking the inode

@@ -784,11 +790,11 @@ out:
  * Note that O_APPEND is not supported for NFS direct writes, as there
  * is no atomic O_APPEND write facility in the NFS protocol.
  */
-ssize_t nfs_file_direct_write(struct kiocb *iocb, const struct iovec *iov,
-				unsigned long nr_segs, loff_t pos)
+ssize_t nfs_file_direct_write(struct file *file, const struct iovec *iov,
+			      unsigned long nr_segs, loff_t *pos,
+			      file_endio_t *endio, void *endio_data)
 {
 	ssize_t retval;
-	struct file *file = iocb->ki_filp;
 	struct address_space *mapping = file->f_mapping;
 	/* XXX: temporary */
 	const char __user *buf = iov[0].iov_base;

@@ -797,12 +803,12 @@ ssize_t nfs_file_direct_write(struct kio
 	dfprintk(VFS, "nfs: direct write(%s/%s, %lu@%Ld)\n",
 		file->f_path.dentry->d_parent->d_name.name,
 		file->f_path.dentry->d_name.name,
-		(unsigned long) count, (long long) pos);
+		(unsigned long) count, (long long) *pos);
 
 	if (nr_segs != 1)
 		return -EINVAL;
 
-	retval = generic_write_checks(file, &pos, &count, 0);
+	retval = generic_write_checks(file, pos, &count, 0);
 	if (retval)
 		goto out;
 

@@ -821,10 +827,11 @@ ssize_t nfs_file_direct_write(struct kio
 	if (retval)
 		goto out;
 
-	retval = nfs_direct_write(iocb, (unsigned long) buf, count, pos);
+	retval = nfs_direct_write(file, (unsigned long) buf, count, *pos,
+				  endio, endio_data);
 
 	if (retval > 0)
-		iocb->ki_pos = pos + retval;
+		*pos += retval;
 
 out:
 	return retval;

@@ -208,7 +208,8 @@ nfs_file_read(struct kiocb *iocb, const 
 
 #ifdef CONFIG_NFS_DIRECTIO
 	if (iocb->ki_filp->f_flags & O_DIRECT)
-		return nfs_file_direct_read(iocb, iov, nr_segs, pos);
+		return nfs_file_direct_read(iocb->ki_filp, iov, nr_segs,
+					    &iocb->ki_pos, aio_complete, iocb);
 #endif
 
 	dfprintk(VFS, "nfs: read(%s/%s, %lu@%lu)\n",

@@ -350,7 +351,8 @@ static ssize_t nfs_file_write(struct kio
 
 #ifdef CONFIG_NFS_DIRECTIO
 	if (iocb->ki_filp->f_flags & O_DIRECT)
-		return nfs_file_direct_write(iocb, iov, nr_segs, pos);
+		return nfs_file_direct_write(iocb->ki_filp, iov, nr_segs,
+					     &iocb->ki_pos, aio_complete, iocb);
 #endif
 
 	dfprintk(VFS, "nfs: write(%s/%s(%ld), %lu@%Ld)\n",

@@ -369,12 +369,12 @@ extern int nfs3_removexattr (struct dent
  */
 extern ssize_t nfs_direct_IO(int, struct kiocb *, const struct iovec *, loff_t,
 			unsigned long);
-extern ssize_t nfs_file_direct_read(struct kiocb *iocb,
+extern ssize_t nfs_file_direct_read(struct file *file,
 			const struct iovec *iov, unsigned long nr_segs,
-			loff_t pos);
-extern ssize_t nfs_file_direct_write(struct kiocb *iocb,
+			loff_t *pos, file_endio_t *endio, void *endio_data);
+extern ssize_t nfs_file_direct_write(struct file *file,
 			const struct iovec *iov, unsigned long nr_segs,
-			loff_t pos);
+			loff_t *pos, file_endio_t *endio, void *endio_data);
 
 /*
  * linux/fs/nfs/dir.c

--

@@ -1157,7 +1157,7 @@ static ssize_t ocfs2_file_aio_write(stru
 		   filp->f_path.dentry->d_name.name);
 
 	/* happy write of zero bytes */
-	if (iocb->ki_left == 0)
+	if (iov_length(iov, nr_segs) == 0)
 		return 0;
 
 	mutex_lock(&inode->i_mutex);

@@ -1177,7 +1177,7 @@ static ssize_t ocfs2_file_aio_write(stru
 	}
 
 	ret = ocfs2_prepare_inode_for_write(filp->f_path.dentry, &iocb->ki_pos,
-					    iocb->ki_left, appending);
+					iov_length(iov, nr_segs), appending);
 	if (ret < 0) {
 		mlog_errno(ret);
 		goto out;