2 * linux/fs/nfs/direct.c
4 * Copyright (C) 2003 by Chuck Lever <cel@netapp.com>
6 * High-performance uncached I/O for the Linux NFS client
8 * There are important applications whose performance or correctness
9 * depends on uncached access to file data. Database clusters
10 * (multiple copies of the same instance running on separate hosts)
11 * implement their own cache coherency protocol that subsumes file
12 * system cache protocols. Applications that process datasets
13 * considerably larger than the client's memory do not always benefit
14 * from a local cache. A streaming video server, for instance, has no
15 * need to cache the contents of a file.
17 * When an application requests uncached I/O, all read and write requests
18 * are made directly to the server; data stored or fetched via these
19 * requests is not cached in the Linux page cache. The client does not
20 * correct unaligned requests from applications. All requested bytes are
21 * held on permanent storage before a direct write system call returns to
24 * Solaris implements an uncached I/O facility called directio() that
25 * is used for backups and sequential I/O to very large files. Solaris
26 * also supports uncaching whole NFS partitions with "-o forcedirectio,"
27 * an undocumented mount option.
29 * Designed by Jeff Kimmel, Chuck Lever, and Trond Myklebust, with
30 * help from Andrew Morton.
32 * 18 Dec 2001 Initial implementation for 2.4 --cel
33 * 08 Jul 2002 Version for 2.4.19, with bug fixes --trondmy
34 * 08 Jun 2003 Port to 2.5 APIs --cel
35 * 31 Mar 2004 Handle direct I/O without VFS support --cel
36 * 15 Sep 2004 Parallel async reads --cel
40 #include <linux/config.h>
41 #include <linux/errno.h>
42 #include <linux/sched.h>
43 #include <linux/kernel.h>
44 #include <linux/smp_lock.h>
45 #include <linux/file.h>
46 #include <linux/pagemap.h>
47 #include <linux/kref.h>
49 #include <linux/nfs_fs.h>
50 #include <linux/nfs_page.h>
51 #include <linux/sunrpc/clnt.h>
53 #include <asm/system.h>
54 #include <asm/uaccess.h>
55 #include <asm/atomic.h>
59 #define NFSDBG_FACILITY NFSDBG_VFS
60 #define MAX_DIRECTIO_SIZE (4096UL << PAGE_SHIFT)
62 static void nfs_free_user_pages(struct page **pages, int npages, int do_dirty);
63 static kmem_cache_t *nfs_direct_cachep;
66 * This represents a set of asynchronous requests that we're waiting on
68 struct nfs_direct_req {
69 struct kref kref; /* release manager */
70 struct list_head list; /* nfs_read_data structs */
71 struct file * filp; /* file descriptor */
72 struct kiocb * iocb; /* controlling i/o request */
73 wait_queue_head_t wait; /* wait for i/o completion */
74 struct inode * inode; /* target file of I/O */
75 struct page ** pages; /* pages in our buffer */
76 unsigned int npages; /* count of pages */
77 atomic_t complete, /* i/os we're waiting for */
78 count, /* bytes actually processed */
79 error; /* any reported error */
84 * nfs_direct_IO - NFS address space operation for direct I/O
85 * @rw: direction (read or write)
86 * @iocb: target I/O control block
87 * @iov: array of vectors that define I/O buffer
88 * @pos: offset in file to begin the operation
89 * @nr_segs: size of iovec array
91 * The presence of this routine in the address space ops vector means
92 * the NFS client supports direct I/O. However, we shunt off direct
93 * read and write requests before the VFS gets them, so this method
94 * should never be called.
96 ssize_t nfs_direct_IO(int rw, struct kiocb *iocb, const struct iovec *iov, loff_t pos, unsigned long nr_segs)
98 struct dentry *dentry = iocb->ki_filp->f_dentry;
100 dprintk("NFS: nfs_direct_IO (%s) off/no(%Ld/%lu) EINVAL\n",
101 dentry->d_name.name, (long long) pos, nr_segs);
106 static inline int nfs_get_user_pages(int rw, unsigned long user_addr, size_t size, struct page ***pages)
108 int result = -ENOMEM;
109 unsigned long page_count;
112 /* set an arbitrary limit to prevent type overflow */
113 /* XXX: this can probably be as large as INT_MAX */
114 if (size > MAX_DIRECTIO_SIZE) {
119 page_count = (user_addr + size + PAGE_SIZE - 1) >> PAGE_SHIFT;
120 page_count -= user_addr >> PAGE_SHIFT;
122 array_size = (page_count * sizeof(struct page *));
123 *pages = kmalloc(array_size, GFP_KERNEL);
125 down_read(¤t->mm->mmap_sem);
126 result = get_user_pages(current, current->mm, user_addr,
127 page_count, (rw == READ), 0,
129 up_read(¤t->mm->mmap_sem);
131 * If we got fewer pages than expected from get_user_pages(),
132 * the user buffer runs off the end of a mapping; return EFAULT.
134 if (result >= 0 && result < page_count) {
135 nfs_free_user_pages(*pages, result, 0);
143 static void nfs_free_user_pages(struct page **pages, int npages, int do_dirty)
146 for (i = 0; i < npages; i++) {
147 struct page *page = pages[i];
148 if (do_dirty && !PageCompound(page))
149 set_page_dirty_lock(page);
150 page_cache_release(page);
155 static inline struct nfs_direct_req *nfs_direct_req_alloc(void)
157 struct nfs_direct_req *dreq;
159 dreq = kmem_cache_alloc(nfs_direct_cachep, SLAB_KERNEL);
163 kref_init(&dreq->kref);
164 init_waitqueue_head(&dreq->wait);
165 INIT_LIST_HEAD(&dreq->list);
167 atomic_set(&dreq->count, 0);
168 atomic_set(&dreq->error, 0);
173 static void nfs_direct_req_release(struct kref *kref)
175 struct nfs_direct_req *dreq = container_of(kref, struct nfs_direct_req, kref);
176 kmem_cache_free(nfs_direct_cachep, dreq);
180 * Collects and returns the final error value/byte-count.
182 static ssize_t nfs_direct_wait(struct nfs_direct_req *dreq)
184 int result = -EIOCBQUEUED;
186 /* Async requests don't wait here */
190 result = wait_event_interruptible(dreq->wait,
191 (atomic_read(&dreq->complete) == 0));
194 result = atomic_read(&dreq->error);
196 result = atomic_read(&dreq->count);
199 kref_put(&dreq->kref, nfs_direct_req_release);
200 return (ssize_t) result;
204 * We must hold a reference to all the pages in this direct read request
205 * until the RPCs complete. This could be long *after* we are woken up in
206 * nfs_direct_wait (for instance, if someone hits ^C on a slow server).
208 * In addition, synchronous I/O uses a stack-allocated iocb. Thus we
209 * can't trust the iocb is still valid here if this is a synchronous
210 * request. If the waiter is woken prematurely, the iocb is long gone.
212 static void nfs_direct_complete(struct nfs_direct_req *dreq)
214 nfs_free_user_pages(dreq->pages, dreq->npages, 1);
217 long res = atomic_read(&dreq->error);
219 res = atomic_read(&dreq->count);
220 aio_complete(dreq->iocb, res, 0);
222 wake_up(&dreq->wait);
224 kref_put(&dreq->kref, nfs_direct_req_release);
228 * Note we also set the number of requests we have in the dreq when we are
229 * done. This prevents races with I/O completion so we will always wait
230 * until all requests have been dispatched and completed.
232 static struct nfs_direct_req *nfs_direct_read_alloc(size_t nbytes, size_t rsize)
234 struct list_head *list;
235 struct nfs_direct_req *dreq;
236 unsigned int reads = 0;
237 unsigned int rpages = (rsize + PAGE_CACHE_SIZE - 1) >> PAGE_CACHE_SHIFT;
239 dreq = nfs_direct_req_alloc();
245 struct nfs_read_data *data = nfs_readdata_alloc(rpages);
247 if (unlikely(!data)) {
248 while (!list_empty(list)) {
249 data = list_entry(list->next,
250 struct nfs_read_data, pages);
251 list_del(&data->pages);
252 nfs_readdata_free(data);
254 kref_put(&dreq->kref, nfs_direct_req_release);
258 INIT_LIST_HEAD(&data->pages);
259 list_add(&data->pages, list);
261 data->req = (struct nfs_page *) dreq;
267 kref_get(&dreq->kref);
268 atomic_set(&dreq->complete, reads);
272 static void nfs_direct_read_result(struct rpc_task *task, void *calldata)
274 struct nfs_read_data *data = calldata;
275 struct nfs_direct_req *dreq = (struct nfs_direct_req *) data->req;
277 if (nfs_readpage_result(task, data) != 0)
279 if (likely(task->tk_status >= 0))
280 atomic_add(data->res.count, &dreq->count);
282 atomic_set(&dreq->error, task->tk_status);
284 if (unlikely(atomic_dec_and_test(&dreq->complete)))
285 nfs_direct_complete(dreq);
288 static const struct rpc_call_ops nfs_read_direct_ops = {
289 .rpc_call_done = nfs_direct_read_result,
290 .rpc_release = nfs_readdata_release,
294 * For each nfs_read_data struct that was allocated on the list, dispatch
295 * an NFS READ operation
297 static void nfs_direct_read_schedule(struct nfs_direct_req *dreq, unsigned long user_addr, size_t count, loff_t file_offset)
299 struct file *file = dreq->filp;
300 struct inode *inode = file->f_mapping->host;
301 struct nfs_open_context *ctx = (struct nfs_open_context *)
303 struct list_head *list = &dreq->list;
304 struct page **pages = dreq->pages;
305 size_t rsize = NFS_SERVER(inode)->rsize;
306 unsigned int curpage, pgbase;
309 pgbase = user_addr & ~PAGE_MASK;
311 struct nfs_read_data *data;
318 data = list_entry(list->next, struct nfs_read_data, pages);
319 list_del_init(&data->pages);
322 data->cred = ctx->cred;
323 data->args.fh = NFS_FH(inode);
324 data->args.context = ctx;
325 data->args.offset = file_offset;
326 data->args.pgbase = pgbase;
327 data->args.pages = &pages[curpage];
328 data->args.count = bytes;
329 data->res.fattr = &data->fattr;
331 data->res.count = bytes;
333 rpc_init_task(&data->task, NFS_CLIENT(inode), RPC_TASK_ASYNC,
334 &nfs_read_direct_ops, data);
335 NFS_PROTO(inode)->read_setup(data);
337 data->task.tk_cookie = (unsigned long) inode;
340 rpc_execute(&data->task);
343 dfprintk(VFS, "NFS: %4d initiated direct read call (req %s/%Ld, %u bytes @ offset %Lu)\n",
346 (long long)NFS_FILEID(inode),
348 (unsigned long long)data->args.offset);
350 file_offset += bytes;
352 curpage += pgbase >> PAGE_SHIFT;
353 pgbase &= ~PAGE_MASK;
356 } while (count != 0);
359 static ssize_t nfs_direct_read(struct kiocb *iocb, unsigned long user_addr, size_t count, loff_t file_offset, struct page **pages, unsigned int nr_pages)
363 struct inode *inode = iocb->ki_filp->f_mapping->host;
364 struct rpc_clnt *clnt = NFS_CLIENT(inode);
365 struct nfs_direct_req *dreq;
367 dreq = nfs_direct_read_alloc(count, NFS_SERVER(inode)->rsize);
372 dreq->npages = nr_pages;
374 dreq->filp = iocb->ki_filp;
375 if (!is_sync_kiocb(iocb))
378 nfs_add_stats(inode, NFSIOS_DIRECTREADBYTES, count);
379 rpc_clnt_sigmask(clnt, &oldset);
380 nfs_direct_read_schedule(dreq, user_addr, count, file_offset);
381 result = nfs_direct_wait(dreq);
382 rpc_clnt_sigunmask(clnt, &oldset);
387 static struct nfs_direct_req *nfs_direct_write_alloc(size_t nbytes, size_t wsize)
389 struct list_head *list;
390 struct nfs_direct_req *dreq;
391 unsigned int writes = 0;
392 unsigned int wpages = (wsize + PAGE_CACHE_SIZE - 1) >> PAGE_CACHE_SHIFT;
394 dreq = nfs_direct_req_alloc();
400 struct nfs_write_data *data = nfs_writedata_alloc(wpages);
402 if (unlikely(!data)) {
403 while (!list_empty(list)) {
404 data = list_entry(list->next,
405 struct nfs_write_data, pages);
406 list_del(&data->pages);
407 nfs_writedata_free(data);
409 kref_put(&dreq->kref, nfs_direct_req_release);
413 INIT_LIST_HEAD(&data->pages);
414 list_add(&data->pages, list);
416 data->req = (struct nfs_page *) dreq;
422 kref_get(&dreq->kref);
423 atomic_set(&dreq->complete, writes);
428 * Collects and returns the final error value/byte-count.
430 static ssize_t nfs_direct_write_wait(struct nfs_direct_req *dreq, int intr)
435 result = wait_event_interruptible(dreq->wait,
436 (atomic_read(&dreq->complete) == 0));
438 wait_event(dreq->wait, (atomic_read(&dreq->complete) == 0));
442 result = atomic_read(&dreq->error);
444 result = atomic_read(&dreq->count);
446 kref_put(&dreq->kref, nfs_direct_req_release);
447 return (ssize_t) result;
450 static void nfs_direct_write_result(struct rpc_task *task, void *calldata)
452 struct nfs_write_data *data = calldata;
453 struct nfs_direct_req *dreq = (struct nfs_direct_req *) data->req;
454 int status = task->tk_status;
456 if (nfs_writeback_done(task, data) != 0)
458 /* If the server fell back to an UNSTABLE write, it's an error. */
459 if (unlikely(data->res.verf->committed != NFS_FILE_SYNC))
462 if (likely(status >= 0))
463 atomic_add(data->res.count, &dreq->count);
465 atomic_set(&dreq->error, status);
467 if (unlikely(atomic_dec_and_test(&dreq->complete)))
468 nfs_direct_complete(dreq);
471 static const struct rpc_call_ops nfs_write_direct_ops = {
472 .rpc_call_done = nfs_direct_write_result,
473 .rpc_release = nfs_writedata_release,
477 * For each nfs_write_data struct that was allocated on the list, dispatch
478 * an NFS WRITE operation
480 * XXX: For now, support only FILE_SYNC writes. Later we may add
481 * support for UNSTABLE + COMMIT.
483 static void nfs_direct_write_schedule(struct nfs_direct_req *dreq, struct inode *inode, struct nfs_open_context *ctx, unsigned long user_addr, size_t count, loff_t file_offset)
485 struct list_head *list = &dreq->list;
486 struct page **pages = dreq->pages;
487 size_t wsize = NFS_SERVER(inode)->wsize;
488 unsigned int curpage, pgbase;
491 pgbase = user_addr & ~PAGE_MASK;
493 struct nfs_write_data *data;
500 data = list_entry(list->next, struct nfs_write_data, pages);
501 list_del_init(&data->pages);
504 data->cred = ctx->cred;
505 data->args.fh = NFS_FH(inode);
506 data->args.context = ctx;
507 data->args.offset = file_offset;
508 data->args.pgbase = pgbase;
509 data->args.pages = &pages[curpage];
510 data->args.count = bytes;
511 data->res.fattr = &data->fattr;
512 data->res.count = bytes;
513 data->res.verf = &data->verf;
515 rpc_init_task(&data->task, NFS_CLIENT(inode), RPC_TASK_ASYNC,
516 &nfs_write_direct_ops, data);
517 NFS_PROTO(inode)->write_setup(data, FLUSH_STABLE);
519 data->task.tk_priority = RPC_PRIORITY_NORMAL;
520 data->task.tk_cookie = (unsigned long) inode;
523 rpc_execute(&data->task);
526 dfprintk(VFS, "NFS: %4d initiated direct write call (req %s/%Ld, %u bytes @ offset %Lu)\n",
529 (long long)NFS_FILEID(inode),
531 (unsigned long long)data->args.offset);
533 file_offset += bytes;
535 curpage += pgbase >> PAGE_SHIFT;
536 pgbase &= ~PAGE_MASK;
539 } while (count != 0);
542 static ssize_t nfs_direct_write(struct inode *inode, struct nfs_open_context *ctx, unsigned long user_addr, size_t count, loff_t file_offset, struct page **pages, int nr_pages)
546 struct rpc_clnt *clnt = NFS_CLIENT(inode);
547 struct nfs_direct_req *dreq;
549 dreq = nfs_direct_write_alloc(count, NFS_SERVER(inode)->wsize);
554 dreq->npages = nr_pages;
556 nfs_add_stats(inode, NFSIOS_DIRECTWRITTENBYTES, count);
558 nfs_begin_data_update(inode);
560 rpc_clnt_sigmask(clnt, &oldset);
561 nfs_direct_write_schedule(dreq, inode, ctx, user_addr, count,
563 result = nfs_direct_write_wait(dreq, clnt->cl_intr);
564 rpc_clnt_sigunmask(clnt, &oldset);
566 nfs_end_data_update(inode);
572 * nfs_file_direct_read - file direct read operation for NFS files
573 * @iocb: target I/O control block
574 * @buf: user's buffer into which to read data
575 * count: number of bytes to read
576 * pos: byte offset in file where reading starts
578 * We use this function for direct reads instead of calling
579 * generic_file_aio_read() in order to avoid gfar's check to see if
580 * the request starts before the end of the file. For that check
581 * to work, we must generate a GETATTR before each direct read, and
582 * even then there is a window between the GETATTR and the subsequent
583 * READ where the file size could change. So our preference is simply
584 * to do all reads the application wants, and the server will take
585 * care of managing the end of file boundary.
587 * This function also eliminates unnecessarily updating the file's
588 * atime locally, as the NFS server sets the file's atime, and this
589 * client must read the updated atime from the server back into its
592 ssize_t nfs_file_direct_read(struct kiocb *iocb, char __user *buf, size_t count, loff_t pos)
594 ssize_t retval = -EINVAL;
597 struct file *file = iocb->ki_filp;
598 struct address_space *mapping = file->f_mapping;
600 dprintk("nfs: direct read(%s/%s, %lu@%Ld)\n",
601 file->f_dentry->d_parent->d_name.name,
602 file->f_dentry->d_name.name,
603 (unsigned long) count, (long long) pos);
608 if (!access_ok(VERIFY_WRITE, buf, count))
614 retval = nfs_sync_mapping(mapping);
618 page_count = nfs_get_user_pages(READ, (unsigned long) buf,
620 if (page_count < 0) {
621 nfs_free_user_pages(pages, 0, 0);
626 retval = nfs_direct_read(iocb, (unsigned long) buf, count, pos,
629 iocb->ki_pos = pos + retval;
636 * nfs_file_direct_write - file direct write operation for NFS files
637 * @iocb: target I/O control block
638 * @buf: user's buffer from which to write data
639 * count: number of bytes to write
640 * pos: byte offset in file where writing starts
642 * We use this function for direct writes instead of calling
643 * generic_file_aio_write() in order to avoid taking the inode
644 * semaphore and updating the i_size. The NFS server will set
645 * the new i_size and this client must read the updated size
646 * back into its cache. We let the server do generic write
647 * parameter checking and report problems.
649 * We also avoid an unnecessary invocation of generic_osync_inode(),
650 * as it is fairly meaningless to sync the metadata of an NFS file.
652 * We eliminate local atime updates, see direct read above.
654 * We avoid unnecessary page cache invalidations for normal cached
655 * readers of this file.
657 * Note that O_APPEND is not supported for NFS direct writes, as there
658 * is no atomic O_APPEND write facility in the NFS protocol.
660 ssize_t nfs_file_direct_write(struct kiocb *iocb, const char __user *buf, size_t count, loff_t pos)
665 struct file *file = iocb->ki_filp;
666 struct nfs_open_context *ctx =
667 (struct nfs_open_context *) file->private_data;
668 struct address_space *mapping = file->f_mapping;
669 struct inode *inode = mapping->host;
671 dfprintk(VFS, "nfs: direct write(%s/%s, %lu@%Ld)\n",
672 file->f_dentry->d_parent->d_name.name,
673 file->f_dentry->d_name.name,
674 (unsigned long) count, (long long) pos);
677 if (!is_sync_kiocb(iocb))
680 retval = generic_write_checks(file, &pos, &count, 0);
685 if ((ssize_t) count < 0)
692 if (!access_ok(VERIFY_READ, buf, count))
695 retval = nfs_sync_mapping(mapping);
699 page_count = nfs_get_user_pages(WRITE, (unsigned long) buf,
701 if (page_count < 0) {
702 nfs_free_user_pages(pages, 0, 0);
707 retval = nfs_direct_write(inode, ctx, (unsigned long) buf, count,
708 pos, pages, page_count);
709 if (mapping->nrpages)
710 invalidate_inode_pages2(mapping);
712 iocb->ki_pos = pos + retval;
718 int nfs_init_directcache(void)
720 nfs_direct_cachep = kmem_cache_create("nfs_direct_cache",
721 sizeof(struct nfs_direct_req),
722 0, SLAB_RECLAIM_ACCOUNT,
724 if (nfs_direct_cachep == NULL)
730 void nfs_destroy_directcache(void)
732 if (kmem_cache_destroy(nfs_direct_cachep))
733 printk(KERN_INFO "nfs_direct_cache: not all structures were freed\n");