[PATCH] dio: only call aio_complete() after returning -EIOCBQUEUED
[linux-2.6.git] / fs / direct-io.c
1 /*
2  * fs/direct-io.c
3  *
4  * Copyright (C) 2002, Linus Torvalds.
5  *
6  * O_DIRECT
7  *
8  * 04Jul2002    akpm@zip.com.au
9  *              Initial version
10  * 11Sep2002    janetinc@us.ibm.com
11  *              added readv/writev support.
12  * 29Oct2002    akpm@zip.com.au
13  *              rewrote bio_add_page() support.
14  * 30Oct2002    pbadari@us.ibm.com
15  *              added support for non-aligned IO.
16  * 06Nov2002    pbadari@us.ibm.com
17  *              added asynchronous IO support.
18  * 21Jul2003    nathans@sgi.com
19  *              added IO completion notifier.
20  */
21
22 #include <linux/kernel.h>
23 #include <linux/module.h>
24 #include <linux/types.h>
25 #include <linux/fs.h>
26 #include <linux/mm.h>
27 #include <linux/slab.h>
28 #include <linux/highmem.h>
29 #include <linux/pagemap.h>
30 #include <linux/task_io_accounting_ops.h>
31 #include <linux/bio.h>
32 #include <linux/wait.h>
33 #include <linux/err.h>
34 #include <linux/blkdev.h>
35 #include <linux/buffer_head.h>
36 #include <linux/rwsem.h>
37 #include <linux/uio.h>
38 #include <asm/atomic.h>
39
40 /*
41  * How many user pages to map in one call to get_user_pages().  This determines
42  * the size of a structure on the stack.
43  */
44 #define DIO_PAGES       64
45
46 /*
47  * This code generally works in units of "dio_blocks".  A dio_block is
48  * somewhere between the hard sector size and the filesystem block size.  it
49  * is determined on a per-invocation basis.   When talking to the filesystem
50  * we need to convert dio_blocks to fs_blocks by scaling the dio_block quantity
51  * down by dio->blkfactor.  Similarly, fs-blocksize quantities are converted
52  * to bio_block quantities by shifting left by blkfactor.
53  *
54  * If blkfactor is zero then the user's request was aligned to the filesystem's
55  * blocksize.
56  *
57  * lock_type is DIO_LOCKING for regular files on direct-IO-naive filesystems.
58  * This determines whether we need to do the fancy locking which prevents
59  * direct-IO from being able to read uninitialised disk blocks.  If its zero
60  * (blockdev) this locking is not done, and if it is DIO_OWN_LOCKING i_mutex is
61  * not held for the entire direct write (taken briefly, initially, during a
62  * direct read though, but its never held for the duration of a direct-IO).
63  */
64
65 struct dio {
66         /* BIO submission state */
67         struct bio *bio;                /* bio under assembly */
68         struct inode *inode;
69         int rw;
70         loff_t i_size;                  /* i_size when submitted */
71         int lock_type;                  /* doesn't change */
72         unsigned blkbits;               /* doesn't change */
73         unsigned blkfactor;             /* When we're using an alignment which
74                                            is finer than the filesystem's soft
75                                            blocksize, this specifies how much
76                                            finer.  blkfactor=2 means 1/4-block
77                                            alignment.  Does not change */
78         unsigned start_zero_done;       /* flag: sub-blocksize zeroing has
79                                            been performed at the start of a
80                                            write */
81         int pages_in_io;                /* approximate total IO pages */
82         size_t  size;                   /* total request size (doesn't change)*/
83         sector_t block_in_file;         /* Current offset into the underlying
84                                            file in dio_block units. */
85         unsigned blocks_available;      /* At block_in_file.  changes */
86         sector_t final_block_in_request;/* doesn't change */
87         unsigned first_block_in_page;   /* doesn't change, Used only once */
88         int boundary;                   /* prev block is at a boundary */
89         int reap_counter;               /* rate limit reaping */
90         get_block_t *get_block;         /* block mapping function */
91         dio_iodone_t *end_io;           /* IO completion function */
92         sector_t final_block_in_bio;    /* current final block in bio + 1 */
93         sector_t next_block_for_io;     /* next block to be put under IO,
94                                            in dio_blocks units */
95         struct buffer_head map_bh;      /* last get_block() result */
96
97         /*
98          * Deferred addition of a page to the dio.  These variables are
99          * private to dio_send_cur_page(), submit_page_section() and
100          * dio_bio_add_page().
101          */
102         struct page *cur_page;          /* The page */
103         unsigned cur_page_offset;       /* Offset into it, in bytes */
104         unsigned cur_page_len;          /* Nr of bytes at cur_page_offset */
105         sector_t cur_page_block;        /* Where it starts */
106
107         /*
108          * Page fetching state. These variables belong to dio_refill_pages().
109          */
110         int curr_page;                  /* changes */
111         int total_pages;                /* doesn't change */
112         unsigned long curr_user_address;/* changes */
113
114         /*
115          * Page queue.  These variables belong to dio_refill_pages() and
116          * dio_get_page().
117          */
118         struct page *pages[DIO_PAGES];  /* page buffer */
119         unsigned head;                  /* next page to process */
120         unsigned tail;                  /* last valid page + 1 */
121         int page_errors;                /* errno from get_user_pages() */
122
123         /* BIO completion state */
124         atomic_t refcount;              /* direct_io_worker() and bios */
125         spinlock_t bio_lock;            /* protects BIO fields below */
126         struct bio *bio_list;           /* singly linked via bi_private */
127         struct task_struct *waiter;     /* waiting task (NULL if none) */
128
129         /* AIO related stuff */
130         struct kiocb *iocb;             /* kiocb */
131         int is_async;                   /* is IO async ? */
132         int io_error;                   /* IO error in completion path */
133         ssize_t result;                 /* IO result */
134 };
135
136 /*
137  * How many pages are in the queue?
138  */
139 static inline unsigned dio_pages_present(struct dio *dio)
140 {
141         return dio->tail - dio->head;
142 }
143
144 /*
145  * Go grab and pin some userspace pages.   Typically we'll get 64 at a time.
146  */
147 static int dio_refill_pages(struct dio *dio)
148 {
149         int ret;
150         int nr_pages;
151
152         nr_pages = min(dio->total_pages - dio->curr_page, DIO_PAGES);
153         down_read(&current->mm->mmap_sem);
154         ret = get_user_pages(
155                 current,                        /* Task for fault acounting */
156                 current->mm,                    /* whose pages? */
157                 dio->curr_user_address,         /* Where from? */
158                 nr_pages,                       /* How many pages? */
159                 dio->rw == READ,                /* Write to memory? */
160                 0,                              /* force (?) */
161                 &dio->pages[0],
162                 NULL);                          /* vmas */
163         up_read(&current->mm->mmap_sem);
164
165         if (ret < 0 && dio->blocks_available && (dio->rw & WRITE)) {
166                 struct page *page = ZERO_PAGE(dio->curr_user_address);
167                 /*
168                  * A memory fault, but the filesystem has some outstanding
169                  * mapped blocks.  We need to use those blocks up to avoid
170                  * leaking stale data in the file.
171                  */
172                 if (dio->page_errors == 0)
173                         dio->page_errors = ret;
174                 page_cache_get(page);
175                 dio->pages[0] = page;
176                 dio->head = 0;
177                 dio->tail = 1;
178                 ret = 0;
179                 goto out;
180         }
181
182         if (ret >= 0) {
183                 dio->curr_user_address += ret * PAGE_SIZE;
184                 dio->curr_page += ret;
185                 dio->head = 0;
186                 dio->tail = ret;
187                 ret = 0;
188         }
189 out:
190         return ret;     
191 }
192
193 /*
194  * Get another userspace page.  Returns an ERR_PTR on error.  Pages are
195  * buffered inside the dio so that we can call get_user_pages() against a
196  * decent number of pages, less frequently.  To provide nicer use of the
197  * L1 cache.
198  */
199 static struct page *dio_get_page(struct dio *dio)
200 {
201         if (dio_pages_present(dio) == 0) {
202                 int ret;
203
204                 ret = dio_refill_pages(dio);
205                 if (ret)
206                         return ERR_PTR(ret);
207                 BUG_ON(dio_pages_present(dio) == 0);
208         }
209         return dio->pages[dio->head++];
210 }
211
212 /**
213  * dio_complete() - called when all DIO BIO I/O has been completed
214  * @offset: the byte offset in the file of the completed operation
215  *
216  * This releases locks as dictated by the locking type, lets interested parties
217  * know that a DIO operation has completed, and calculates the resulting return
218  * code for the operation.
219  *
220  * It lets the filesystem know if it registered an interest earlier via
221  * get_block.  Pass the private field of the map buffer_head so that
222  * filesystems can use it to hold additional state between get_block calls and
223  * dio_complete.
224  */
225 static int dio_complete(struct dio *dio, loff_t offset, int ret)
226 {
227         ssize_t transferred = 0;
228
229         /*
230          * AIO submission can race with bio completion to get here while
231          * expecting to have the last io completed by bio completion.
232          * In that case -EIOCBQUEUED is in fact not an error we want
233          * to preserve through this call.
234          */
235         if (ret == -EIOCBQUEUED)
236                 ret = 0;
237
238         if (dio->result) {
239                 transferred = dio->result;
240
241                 /* Check for short read case */
242                 if ((dio->rw == READ) && ((offset + transferred) > dio->i_size))
243                         transferred = dio->i_size - offset;
244         }
245
246         if (dio->end_io && dio->result)
247                 dio->end_io(dio->iocb, offset, transferred,
248                             dio->map_bh.b_private);
249         if (dio->lock_type == DIO_LOCKING)
250                 /* lockdep: non-owner release */
251                 up_read_non_owner(&dio->inode->i_alloc_sem);
252
253         if (ret == 0)
254                 ret = dio->page_errors;
255         if (ret == 0)
256                 ret = dio->io_error;
257         if (ret == 0)
258                 ret = transferred;
259
260         return ret;
261 }
262
263 static int dio_bio_complete(struct dio *dio, struct bio *bio);
264 /*
265  * Asynchronous IO callback. 
266  */
267 static int dio_bio_end_aio(struct bio *bio, unsigned int bytes_done, int error)
268 {
269         struct dio *dio = bio->bi_private;
270         int waiter_holds_ref = 0;
271         int remaining;
272
273         if (bio->bi_size)
274                 return 1;
275
276         /* cleanup the bio */
277         dio_bio_complete(dio, bio);
278
279         waiter_holds_ref = !!dio->waiter;
280         remaining = atomic_sub_return(1, (&dio->refcount));
281         if (remaining == 1 && waiter_holds_ref)
282                 wake_up_process(dio->waiter);
283
284         if (remaining == 0) {
285                 int ret = dio_complete(dio, dio->iocb->ki_pos, 0);
286                 aio_complete(dio->iocb, ret, 0);
287                 kfree(dio);
288         }
289
290         return 0;
291 }
292
293 /*
294  * The BIO completion handler simply queues the BIO up for the process-context
295  * handler.
296  *
297  * During I/O bi_private points at the dio.  After I/O, bi_private is used to
298  * implement a singly-linked list of completed BIOs, at dio->bio_list.
299  */
300 static int dio_bio_end_io(struct bio *bio, unsigned int bytes_done, int error)
301 {
302         struct dio *dio = bio->bi_private;
303         unsigned long flags;
304
305         if (bio->bi_size)
306                 return 1;
307
308         spin_lock_irqsave(&dio->bio_lock, flags);
309         bio->bi_private = dio->bio_list;
310         dio->bio_list = bio;
311         if ((atomic_sub_return(1, &dio->refcount) == 1) && dio->waiter)
312                 wake_up_process(dio->waiter);
313         spin_unlock_irqrestore(&dio->bio_lock, flags);
314         return 0;
315 }
316
317 static int
318 dio_bio_alloc(struct dio *dio, struct block_device *bdev,
319                 sector_t first_sector, int nr_vecs)
320 {
321         struct bio *bio;
322
323         bio = bio_alloc(GFP_KERNEL, nr_vecs);
324         if (bio == NULL)
325                 return -ENOMEM;
326
327         bio->bi_bdev = bdev;
328         bio->bi_sector = first_sector;
329         if (dio->is_async)
330                 bio->bi_end_io = dio_bio_end_aio;
331         else
332                 bio->bi_end_io = dio_bio_end_io;
333
334         dio->bio = bio;
335         return 0;
336 }
337
338 /*
339  * In the AIO read case we speculatively dirty the pages before starting IO.
340  * During IO completion, any of these pages which happen to have been written
341  * back will be redirtied by bio_check_pages_dirty().
342  *
343  * bios hold a dio reference between submit_bio and ->end_io.
344  */
345 static void dio_bio_submit(struct dio *dio)
346 {
347         struct bio *bio = dio->bio;
348
349         bio->bi_private = dio;
350         atomic_inc(&dio->refcount);
351         if (dio->is_async && dio->rw == READ)
352                 bio_set_pages_dirty(bio);
353         submit_bio(dio->rw, bio);
354
355         dio->bio = NULL;
356         dio->boundary = 0;
357 }
358
359 /*
360  * Release any resources in case of a failure
361  */
362 static void dio_cleanup(struct dio *dio)
363 {
364         while (dio_pages_present(dio))
365                 page_cache_release(dio_get_page(dio));
366 }
367
368 static int wait_for_more_bios(struct dio *dio)
369 {
370         assert_spin_locked(&dio->bio_lock);
371
372         return (atomic_read(&dio->refcount) > 1) && (dio->bio_list == NULL);
373 }
374
375 /*
376  * Wait for the next BIO to complete.  Remove it and return it.  NULL is
377  * returned once all BIOs have been completed.  This must only be called once
378  * all bios have been issued so that dio->refcount can only decrease.  This
379  * requires that that the caller hold a reference on the dio.
380  */
381 static struct bio *dio_await_one(struct dio *dio)
382 {
383         unsigned long flags;
384         struct bio *bio = NULL;
385
386         spin_lock_irqsave(&dio->bio_lock, flags);
387         while (wait_for_more_bios(dio)) {
388                 set_current_state(TASK_UNINTERRUPTIBLE);
389                 if (wait_for_more_bios(dio)) {
390                         dio->waiter = current;
391                         spin_unlock_irqrestore(&dio->bio_lock, flags);
392                         io_schedule();
393                         spin_lock_irqsave(&dio->bio_lock, flags);
394                         dio->waiter = NULL;
395                 }
396                 set_current_state(TASK_RUNNING);
397         }
398         if (dio->bio_list) {
399                 bio = dio->bio_list;
400                 dio->bio_list = bio->bi_private;
401         }
402         spin_unlock_irqrestore(&dio->bio_lock, flags);
403         return bio;
404 }
405
406 /*
407  * Process one completed BIO.  No locks are held.
408  */
409 static int dio_bio_complete(struct dio *dio, struct bio *bio)
410 {
411         const int uptodate = test_bit(BIO_UPTODATE, &bio->bi_flags);
412         struct bio_vec *bvec = bio->bi_io_vec;
413         int page_no;
414
415         if (!uptodate)
416                 dio->io_error = -EIO;
417
418         if (dio->is_async && dio->rw == READ) {
419                 bio_check_pages_dirty(bio);     /* transfers ownership */
420         } else {
421                 for (page_no = 0; page_no < bio->bi_vcnt; page_no++) {
422                         struct page *page = bvec[page_no].bv_page;
423
424                         if (dio->rw == READ && !PageCompound(page))
425                                 set_page_dirty_lock(page);
426                         page_cache_release(page);
427                 }
428                 bio_put(bio);
429         }
430         return uptodate ? 0 : -EIO;
431 }
432
433 /*
434  * Wait on and process all in-flight BIOs.  This must only be called once
435  * all bios have been issued so that the refcount can only decrease.
436  * This just waits for all bios to make it through dio_bio_complete.  IO
437  * errors are propogated through dio->io_error and should be propogated via
438  * dio_complete().
439  */
440 static void dio_await_completion(struct dio *dio)
441 {
442         struct bio *bio;
443         do {
444                 bio = dio_await_one(dio);
445                 if (bio)
446                         dio_bio_complete(dio, bio);
447         } while (bio);
448 }
449
450 /*
451  * A really large O_DIRECT read or write can generate a lot of BIOs.  So
452  * to keep the memory consumption sane we periodically reap any completed BIOs
453  * during the BIO generation phase.
454  *
455  * This also helps to limit the peak amount of pinned userspace memory.
456  */
457 static int dio_bio_reap(struct dio *dio)
458 {
459         int ret = 0;
460
461         if (dio->reap_counter++ >= 64) {
462                 while (dio->bio_list) {
463                         unsigned long flags;
464                         struct bio *bio;
465                         int ret2;
466
467                         spin_lock_irqsave(&dio->bio_lock, flags);
468                         bio = dio->bio_list;
469                         dio->bio_list = bio->bi_private;
470                         spin_unlock_irqrestore(&dio->bio_lock, flags);
471                         ret2 = dio_bio_complete(dio, bio);
472                         if (ret == 0)
473                                 ret = ret2;
474                 }
475                 dio->reap_counter = 0;
476         }
477         return ret;
478 }
479
480 /*
481  * Call into the fs to map some more disk blocks.  We record the current number
482  * of available blocks at dio->blocks_available.  These are in units of the
483  * fs blocksize, (1 << inode->i_blkbits).
484  *
485  * The fs is allowed to map lots of blocks at once.  If it wants to do that,
486  * it uses the passed inode-relative block number as the file offset, as usual.
487  *
488  * get_block() is passed the number of i_blkbits-sized blocks which direct_io
489  * has remaining to do.  The fs should not map more than this number of blocks.
490  *
491  * If the fs has mapped a lot of blocks, it should populate bh->b_size to
492  * indicate how much contiguous disk space has been made available at
493  * bh->b_blocknr.
494  *
495  * If *any* of the mapped blocks are new, then the fs must set buffer_new().
496  * This isn't very efficient...
497  *
498  * In the case of filesystem holes: the fs may return an arbitrarily-large
499  * hole by returning an appropriate value in b_size and by clearing
500  * buffer_mapped().  However the direct-io code will only process holes one
501  * block at a time - it will repeatedly call get_block() as it walks the hole.
502  */
503 static int get_more_blocks(struct dio *dio)
504 {
505         int ret;
506         struct buffer_head *map_bh = &dio->map_bh;
507         sector_t fs_startblk;   /* Into file, in filesystem-sized blocks */
508         unsigned long fs_count; /* Number of filesystem-sized blocks */
509         unsigned long dio_count;/* Number of dio_block-sized blocks */
510         unsigned long blkmask;
511         int create;
512
513         /*
514          * If there was a memory error and we've overwritten all the
515          * mapped blocks then we can now return that memory error
516          */
517         ret = dio->page_errors;
518         if (ret == 0) {
519                 BUG_ON(dio->block_in_file >= dio->final_block_in_request);
520                 fs_startblk = dio->block_in_file >> dio->blkfactor;
521                 dio_count = dio->final_block_in_request - dio->block_in_file;
522                 fs_count = dio_count >> dio->blkfactor;
523                 blkmask = (1 << dio->blkfactor) - 1;
524                 if (dio_count & blkmask)        
525                         fs_count++;
526
527                 map_bh->b_state = 0;
528                 map_bh->b_size = fs_count << dio->inode->i_blkbits;
529
530                 create = dio->rw & WRITE;
531                 if (dio->lock_type == DIO_LOCKING) {
532                         if (dio->block_in_file < (i_size_read(dio->inode) >>
533                                                         dio->blkbits))
534                                 create = 0;
535                 } else if (dio->lock_type == DIO_NO_LOCKING) {
536                         create = 0;
537                 }
538
539                 /*
540                  * For writes inside i_size we forbid block creations: only
541                  * overwrites are permitted.  We fall back to buffered writes
542                  * at a higher level for inside-i_size block-instantiating
543                  * writes.
544                  */
545                 ret = (*dio->get_block)(dio->inode, fs_startblk,
546                                                 map_bh, create);
547         }
548         return ret;
549 }
550
551 /*
552  * There is no bio.  Make one now.
553  */
554 static int dio_new_bio(struct dio *dio, sector_t start_sector)
555 {
556         sector_t sector;
557         int ret, nr_pages;
558
559         ret = dio_bio_reap(dio);
560         if (ret)
561                 goto out;
562         sector = start_sector << (dio->blkbits - 9);
563         nr_pages = min(dio->pages_in_io, bio_get_nr_vecs(dio->map_bh.b_bdev));
564         BUG_ON(nr_pages <= 0);
565         ret = dio_bio_alloc(dio, dio->map_bh.b_bdev, sector, nr_pages);
566         dio->boundary = 0;
567 out:
568         return ret;
569 }
570
571 /*
572  * Attempt to put the current chunk of 'cur_page' into the current BIO.  If
573  * that was successful then update final_block_in_bio and take a ref against
574  * the just-added page.
575  *
576  * Return zero on success.  Non-zero means the caller needs to start a new BIO.
577  */
578 static int dio_bio_add_page(struct dio *dio)
579 {
580         int ret;
581
582         ret = bio_add_page(dio->bio, dio->cur_page,
583                         dio->cur_page_len, dio->cur_page_offset);
584         if (ret == dio->cur_page_len) {
585                 /*
586                  * Decrement count only, if we are done with this page
587                  */
588                 if ((dio->cur_page_len + dio->cur_page_offset) == PAGE_SIZE)
589                         dio->pages_in_io--;
590                 page_cache_get(dio->cur_page);
591                 dio->final_block_in_bio = dio->cur_page_block +
592                         (dio->cur_page_len >> dio->blkbits);
593                 ret = 0;
594         } else {
595                 ret = 1;
596         }
597         return ret;
598 }
599                 
600 /*
601  * Put cur_page under IO.  The section of cur_page which is described by
602  * cur_page_offset,cur_page_len is put into a BIO.  The section of cur_page
603  * starts on-disk at cur_page_block.
604  *
605  * We take a ref against the page here (on behalf of its presence in the bio).
606  *
607  * The caller of this function is responsible for removing cur_page from the
608  * dio, and for dropping the refcount which came from that presence.
609  */
610 static int dio_send_cur_page(struct dio *dio)
611 {
612         int ret = 0;
613
614         if (dio->bio) {
615                 /*
616                  * See whether this new request is contiguous with the old
617                  */
618                 if (dio->final_block_in_bio != dio->cur_page_block)
619                         dio_bio_submit(dio);
620                 /*
621                  * Submit now if the underlying fs is about to perform a
622                  * metadata read
623                  */
624                 if (dio->boundary)
625                         dio_bio_submit(dio);
626         }
627
628         if (dio->bio == NULL) {
629                 ret = dio_new_bio(dio, dio->cur_page_block);
630                 if (ret)
631                         goto out;
632         }
633
634         if (dio_bio_add_page(dio) != 0) {
635                 dio_bio_submit(dio);
636                 ret = dio_new_bio(dio, dio->cur_page_block);
637                 if (ret == 0) {
638                         ret = dio_bio_add_page(dio);
639                         BUG_ON(ret != 0);
640                 }
641         }
642 out:
643         return ret;
644 }
645
646 /*
647  * An autonomous function to put a chunk of a page under deferred IO.
648  *
649  * The caller doesn't actually know (or care) whether this piece of page is in
650  * a BIO, or is under IO or whatever.  We just take care of all possible 
651  * situations here.  The separation between the logic of do_direct_IO() and
652  * that of submit_page_section() is important for clarity.  Please don't break.
653  *
654  * The chunk of page starts on-disk at blocknr.
655  *
656  * We perform deferred IO, by recording the last-submitted page inside our
657  * private part of the dio structure.  If possible, we just expand the IO
658  * across that page here.
659  *
660  * If that doesn't work out then we put the old page into the bio and add this
661  * page to the dio instead.
662  */
663 static int
664 submit_page_section(struct dio *dio, struct page *page,
665                 unsigned offset, unsigned len, sector_t blocknr)
666 {
667         int ret = 0;
668
669         if (dio->rw & WRITE) {
670                 /*
671                  * Read accounting is performed in submit_bio()
672                  */
673                 task_io_account_write(len);
674         }
675
676         /*
677          * Can we just grow the current page's presence in the dio?
678          */
679         if (    (dio->cur_page == page) &&
680                 (dio->cur_page_offset + dio->cur_page_len == offset) &&
681                 (dio->cur_page_block +
682                         (dio->cur_page_len >> dio->blkbits) == blocknr)) {
683                 dio->cur_page_len += len;
684
685                 /*
686                  * If dio->boundary then we want to schedule the IO now to
687                  * avoid metadata seeks.
688                  */
689                 if (dio->boundary) {
690                         ret = dio_send_cur_page(dio);
691                         page_cache_release(dio->cur_page);
692                         dio->cur_page = NULL;
693                 }
694                 goto out;
695         }
696
697         /*
698          * If there's a deferred page already there then send it.
699          */
700         if (dio->cur_page) {
701                 ret = dio_send_cur_page(dio);
702                 page_cache_release(dio->cur_page);
703                 dio->cur_page = NULL;
704                 if (ret)
705                         goto out;
706         }
707
708         page_cache_get(page);           /* It is in dio */
709         dio->cur_page = page;
710         dio->cur_page_offset = offset;
711         dio->cur_page_len = len;
712         dio->cur_page_block = blocknr;
713 out:
714         return ret;
715 }
716
717 /*
718  * Clean any dirty buffers in the blockdev mapping which alias newly-created
719  * file blocks.  Only called for S_ISREG files - blockdevs do not set
720  * buffer_new
721  */
722 static void clean_blockdev_aliases(struct dio *dio)
723 {
724         unsigned i;
725         unsigned nblocks;
726
727         nblocks = dio->map_bh.b_size >> dio->inode->i_blkbits;
728
729         for (i = 0; i < nblocks; i++) {
730                 unmap_underlying_metadata(dio->map_bh.b_bdev,
731                                         dio->map_bh.b_blocknr + i);
732         }
733 }
734
735 /*
736  * If we are not writing the entire block and get_block() allocated
737  * the block for us, we need to fill-in the unused portion of the
738  * block with zeros. This happens only if user-buffer, fileoffset or
739  * io length is not filesystem block-size multiple.
740  *
741  * `end' is zero if we're doing the start of the IO, 1 at the end of the
742  * IO.
743  */
744 static void dio_zero_block(struct dio *dio, int end)
745 {
746         unsigned dio_blocks_per_fs_block;
747         unsigned this_chunk_blocks;     /* In dio_blocks */
748         unsigned this_chunk_bytes;
749         struct page *page;
750
751         dio->start_zero_done = 1;
752         if (!dio->blkfactor || !buffer_new(&dio->map_bh))
753                 return;
754
755         dio_blocks_per_fs_block = 1 << dio->blkfactor;
756         this_chunk_blocks = dio->block_in_file & (dio_blocks_per_fs_block - 1);
757
758         if (!this_chunk_blocks)
759                 return;
760
761         /*
762          * We need to zero out part of an fs block.  It is either at the
763          * beginning or the end of the fs block.
764          */
765         if (end) 
766                 this_chunk_blocks = dio_blocks_per_fs_block - this_chunk_blocks;
767
768         this_chunk_bytes = this_chunk_blocks << dio->blkbits;
769
770         page = ZERO_PAGE(dio->curr_user_address);
771         if (submit_page_section(dio, page, 0, this_chunk_bytes, 
772                                 dio->next_block_for_io))
773                 return;
774
775         dio->next_block_for_io += this_chunk_blocks;
776 }
777
778 /*
779  * Walk the user pages, and the file, mapping blocks to disk and generating
780  * a sequence of (page,offset,len,block) mappings.  These mappings are injected
781  * into submit_page_section(), which takes care of the next stage of submission
782  *
783  * Direct IO against a blockdev is different from a file.  Because we can
784  * happily perform page-sized but 512-byte aligned IOs.  It is important that
785  * blockdev IO be able to have fine alignment and large sizes.
786  *
787  * So what we do is to permit the ->get_block function to populate bh.b_size
788  * with the size of IO which is permitted at this offset and this i_blkbits.
789  *
790  * For best results, the blockdev should be set up with 512-byte i_blkbits and
791  * it should set b_size to PAGE_SIZE or more inside get_block().  This gives
792  * fine alignment but still allows this function to work in PAGE_SIZE units.
793  */
794 static int do_direct_IO(struct dio *dio)
795 {
796         const unsigned blkbits = dio->blkbits;
797         const unsigned blocks_per_page = PAGE_SIZE >> blkbits;
798         struct page *page;
799         unsigned block_in_page;
800         struct buffer_head *map_bh = &dio->map_bh;
801         int ret = 0;
802
803         /* The I/O can start at any block offset within the first page */
804         block_in_page = dio->first_block_in_page;
805
806         while (dio->block_in_file < dio->final_block_in_request) {
807                 page = dio_get_page(dio);
808                 if (IS_ERR(page)) {
809                         ret = PTR_ERR(page);
810                         goto out;
811                 }
812
813                 while (block_in_page < blocks_per_page) {
814                         unsigned offset_in_page = block_in_page << blkbits;
815                         unsigned this_chunk_bytes;      /* # of bytes mapped */
816                         unsigned this_chunk_blocks;     /* # of blocks */
817                         unsigned u;
818
819                         if (dio->blocks_available == 0) {
820                                 /*
821                                  * Need to go and map some more disk
822                                  */
823                                 unsigned long blkmask;
824                                 unsigned long dio_remainder;
825
826                                 ret = get_more_blocks(dio);
827                                 if (ret) {
828                                         page_cache_release(page);
829                                         goto out;
830                                 }
831                                 if (!buffer_mapped(map_bh))
832                                         goto do_holes;
833
834                                 dio->blocks_available =
835                                                 map_bh->b_size >> dio->blkbits;
836                                 dio->next_block_for_io =
837                                         map_bh->b_blocknr << dio->blkfactor;
838                                 if (buffer_new(map_bh))
839                                         clean_blockdev_aliases(dio);
840
841                                 if (!dio->blkfactor)
842                                         goto do_holes;
843
844                                 blkmask = (1 << dio->blkfactor) - 1;
845                                 dio_remainder = (dio->block_in_file & blkmask);
846
847                                 /*
848                                  * If we are at the start of IO and that IO
849                                  * starts partway into a fs-block,
850                                  * dio_remainder will be non-zero.  If the IO
851                                  * is a read then we can simply advance the IO
852                                  * cursor to the first block which is to be
853                                  * read.  But if the IO is a write and the
854                                  * block was newly allocated we cannot do that;
855                                  * the start of the fs block must be zeroed out
856                                  * on-disk
857                                  */
858                                 if (!buffer_new(map_bh))
859                                         dio->next_block_for_io += dio_remainder;
860                                 dio->blocks_available -= dio_remainder;
861                         }
862 do_holes:
863                         /* Handle holes */
864                         if (!buffer_mapped(map_bh)) {
865                                 char *kaddr;
866                                 loff_t i_size_aligned;
867
868                                 /* AKPM: eargh, -ENOTBLK is a hack */
869                                 if (dio->rw & WRITE) {
870                                         page_cache_release(page);
871                                         return -ENOTBLK;
872                                 }
873
874                                 /*
875                                  * Be sure to account for a partial block as the
876                                  * last block in the file
877                                  */
878                                 i_size_aligned = ALIGN(i_size_read(dio->inode),
879                                                         1 << blkbits);
880                                 if (dio->block_in_file >=
881                                                 i_size_aligned >> blkbits) {
882                                         /* We hit eof */
883                                         page_cache_release(page);
884                                         goto out;
885                                 }
886                                 kaddr = kmap_atomic(page, KM_USER0);
887                                 memset(kaddr + (block_in_page << blkbits),
888                                                 0, 1 << blkbits);
889                                 flush_dcache_page(page);
890                                 kunmap_atomic(kaddr, KM_USER0);
891                                 dio->block_in_file++;
892                                 block_in_page++;
893                                 goto next_block;
894                         }
895
896                         /*
897                          * If we're performing IO which has an alignment which
898                          * is finer than the underlying fs, go check to see if
899                          * we must zero out the start of this block.
900                          */
901                         if (unlikely(dio->blkfactor && !dio->start_zero_done))
902                                 dio_zero_block(dio, 0);
903
904                         /*
905                          * Work out, in this_chunk_blocks, how much disk we
906                          * can add to this page
907                          */
908                         this_chunk_blocks = dio->blocks_available;
909                         u = (PAGE_SIZE - offset_in_page) >> blkbits;
910                         if (this_chunk_blocks > u)
911                                 this_chunk_blocks = u;
912                         u = dio->final_block_in_request - dio->block_in_file;
913                         if (this_chunk_blocks > u)
914                                 this_chunk_blocks = u;
915                         this_chunk_bytes = this_chunk_blocks << blkbits;
916                         BUG_ON(this_chunk_bytes == 0);
917
918                         dio->boundary = buffer_boundary(map_bh);
919                         ret = submit_page_section(dio, page, offset_in_page,
920                                 this_chunk_bytes, dio->next_block_for_io);
921                         if (ret) {
922                                 page_cache_release(page);
923                                 goto out;
924                         }
925                         dio->next_block_for_io += this_chunk_blocks;
926
927                         dio->block_in_file += this_chunk_blocks;
928                         block_in_page += this_chunk_blocks;
929                         dio->blocks_available -= this_chunk_blocks;
930 next_block:
931                         BUG_ON(dio->block_in_file > dio->final_block_in_request);
932                         if (dio->block_in_file == dio->final_block_in_request)
933                                 break;
934                 }
935
936                 /* Drop the ref which was taken in get_user_pages() */
937                 page_cache_release(page);
938                 block_in_page = 0;
939         }
940 out:
941         return ret;
942 }
943
944 /*
945  * Releases both i_mutex and i_alloc_sem
946  */
947 static ssize_t
948 direct_io_worker(int rw, struct kiocb *iocb, struct inode *inode, 
949         const struct iovec *iov, loff_t offset, unsigned long nr_segs, 
950         unsigned blkbits, get_block_t get_block, dio_iodone_t end_io,
951         struct dio *dio)
952 {
953         unsigned long user_addr; 
954         int seg;
955         ssize_t ret = 0;
956         ssize_t ret2;
957         size_t bytes;
958
959         dio->bio = NULL;
960         dio->inode = inode;
961         dio->rw = rw;
962         dio->blkbits = blkbits;
963         dio->blkfactor = inode->i_blkbits - blkbits;
964         dio->start_zero_done = 0;
965         dio->size = 0;
966         dio->block_in_file = offset >> blkbits;
967         dio->blocks_available = 0;
968         dio->cur_page = NULL;
969
970         dio->boundary = 0;
971         dio->reap_counter = 0;
972         dio->get_block = get_block;
973         dio->end_io = end_io;
974         dio->map_bh.b_private = NULL;
975         dio->final_block_in_bio = -1;
976         dio->next_block_for_io = -1;
977
978         dio->page_errors = 0;
979         dio->io_error = 0;
980         dio->result = 0;
981         dio->iocb = iocb;
982         dio->i_size = i_size_read(inode);
983
984         atomic_set(&dio->refcount, 1);
985         spin_lock_init(&dio->bio_lock);
986         dio->bio_list = NULL;
987         dio->waiter = NULL;
988
989         /*
990          * In case of non-aligned buffers, we may need 2 more
991          * pages since we need to zero out first and last block.
992          */
993         if (unlikely(dio->blkfactor))
994                 dio->pages_in_io = 2;
995         else
996                 dio->pages_in_io = 0;
997
998         for (seg = 0; seg < nr_segs; seg++) {
999                 user_addr = (unsigned long)iov[seg].iov_base;
1000                 dio->pages_in_io +=
1001                         ((user_addr+iov[seg].iov_len +PAGE_SIZE-1)/PAGE_SIZE
1002                                 - user_addr/PAGE_SIZE);
1003         }
1004
1005         for (seg = 0; seg < nr_segs; seg++) {
1006                 user_addr = (unsigned long)iov[seg].iov_base;
1007                 dio->size += bytes = iov[seg].iov_len;
1008
1009                 /* Index into the first page of the first block */
1010                 dio->first_block_in_page = (user_addr & ~PAGE_MASK) >> blkbits;
1011                 dio->final_block_in_request = dio->block_in_file +
1012                                                 (bytes >> blkbits);
1013                 /* Page fetching state */
1014                 dio->head = 0;
1015                 dio->tail = 0;
1016                 dio->curr_page = 0;
1017
1018                 dio->total_pages = 0;
1019                 if (user_addr & (PAGE_SIZE-1)) {
1020                         dio->total_pages++;
1021                         bytes -= PAGE_SIZE - (user_addr & (PAGE_SIZE - 1));
1022                 }
1023                 dio->total_pages += (bytes + PAGE_SIZE - 1) / PAGE_SIZE;
1024                 dio->curr_user_address = user_addr;
1025         
1026                 ret = do_direct_IO(dio);
1027
1028                 dio->result += iov[seg].iov_len -
1029                         ((dio->final_block_in_request - dio->block_in_file) <<
1030                                         blkbits);
1031
1032                 if (ret) {
1033                         dio_cleanup(dio);
1034                         break;
1035                 }
1036         } /* end iovec loop */
1037
1038         if (ret == -ENOTBLK && (rw & WRITE)) {
1039                 /*
1040                  * The remaining part of the request will be
1041                  * be handled by buffered I/O when we return
1042                  */
1043                 ret = 0;
1044         }
1045         /*
1046          * There may be some unwritten disk at the end of a part-written
1047          * fs-block-sized block.  Go zero that now.
1048          */
1049         dio_zero_block(dio, 1);
1050
1051         if (dio->cur_page) {
1052                 ret2 = dio_send_cur_page(dio);
1053                 if (ret == 0)
1054                         ret = ret2;
1055                 page_cache_release(dio->cur_page);
1056                 dio->cur_page = NULL;
1057         }
1058         if (dio->bio)
1059                 dio_bio_submit(dio);
1060
1061         /* All IO is now issued, send it on its way */
1062         blk_run_address_space(inode->i_mapping);
1063
1064         /*
1065          * It is possible that, we return short IO due to end of file.
1066          * In that case, we need to release all the pages we got hold on.
1067          */
1068         dio_cleanup(dio);
1069
1070         /*
1071          * All block lookups have been performed. For READ requests
1072          * we can let i_mutex go now that its achieved its purpose
1073          * of protecting us from looking up uninitialized blocks.
1074          */
1075         if ((rw == READ) && (dio->lock_type == DIO_LOCKING))
1076                 mutex_unlock(&dio->inode->i_mutex);
1077
1078         /*
1079          * The only time we want to leave bios in flight is when a successful
1080          * partial aio read or full aio write have been setup.  In that case
1081          * bio completion will call aio_complete.  The only time it's safe to
1082          * call aio_complete is when we return -EIOCBQUEUED, so we key on that.
1083          * This had *better* be the only place that raises -EIOCBQUEUED.
1084          */
1085         BUG_ON(ret == -EIOCBQUEUED);
1086         if (dio->is_async && ret == 0 && dio->result &&
1087             ((rw & READ) || (dio->result == dio->size)))
1088                 ret = -EIOCBQUEUED;
1089
1090         if (ret != -EIOCBQUEUED)
1091                 dio_await_completion(dio);
1092
1093         /*
1094          * Sync will always be dropping the final ref and completing the
1095          * operation.  AIO can if it was a broken operation described above
1096          * or in fact if all the bios race to complete before we get here.
1097          * In that case dio_complete() translates the EIOCBQUEUED into
1098          * the proper return code that the caller will hand to aio_complete().
1099          */
1100         if (atomic_dec_and_test(&dio->refcount)) {
1101                 ret = dio_complete(dio, offset, ret);
1102                 kfree(dio);
1103         } else
1104                 BUG_ON(ret != -EIOCBQUEUED);
1105
1106         return ret;
1107 }
1108
1109 /*
1110  * This is a library function for use by filesystem drivers.
1111  * The locking rules are governed by the dio_lock_type parameter.
1112  *
1113  * DIO_NO_LOCKING (no locking, for raw block device access)
1114  * For writes, i_mutex is not held on entry; it is never taken.
1115  *
1116  * DIO_LOCKING (simple locking for regular files)
1117  * For writes we are called under i_mutex and return with i_mutex held, even
1118  * though it is internally dropped.
1119  * For reads, i_mutex is not held on entry, but it is taken and dropped before
1120  * returning.
1121  *
1122  * DIO_OWN_LOCKING (filesystem provides synchronisation and handling of
1123  *      uninitialised data, allowing parallel direct readers and writers)
1124  * For writes we are called without i_mutex, return without it, never touch it.
1125  * For reads we are called under i_mutex and return with i_mutex held, even
1126  * though it may be internally dropped.
1127  *
1128  * Additional i_alloc_sem locking requirements described inline below.
1129  */
1130 ssize_t
1131 __blockdev_direct_IO(int rw, struct kiocb *iocb, struct inode *inode,
1132         struct block_device *bdev, const struct iovec *iov, loff_t offset, 
1133         unsigned long nr_segs, get_block_t get_block, dio_iodone_t end_io,
1134         int dio_lock_type)
1135 {
1136         int seg;
1137         size_t size;
1138         unsigned long addr;
1139         unsigned blkbits = inode->i_blkbits;
1140         unsigned bdev_blkbits = 0;
1141         unsigned blocksize_mask = (1 << blkbits) - 1;
1142         ssize_t retval = -EINVAL;
1143         loff_t end = offset;
1144         struct dio *dio;
1145         int release_i_mutex = 0;
1146         int acquire_i_mutex = 0;
1147
1148         if (rw & WRITE)
1149                 rw = WRITE_SYNC;
1150
1151         if (bdev)
1152                 bdev_blkbits = blksize_bits(bdev_hardsect_size(bdev));
1153
1154         if (offset & blocksize_mask) {
1155                 if (bdev)
1156                          blkbits = bdev_blkbits;
1157                 blocksize_mask = (1 << blkbits) - 1;
1158                 if (offset & blocksize_mask)
1159                         goto out;
1160         }
1161
1162         /* Check the memory alignment.  Blocks cannot straddle pages */
1163         for (seg = 0; seg < nr_segs; seg++) {
1164                 addr = (unsigned long)iov[seg].iov_base;
1165                 size = iov[seg].iov_len;
1166                 end += size;
1167                 if ((addr & blocksize_mask) || (size & blocksize_mask))  {
1168                         if (bdev)
1169                                  blkbits = bdev_blkbits;
1170                         blocksize_mask = (1 << blkbits) - 1;
1171                         if ((addr & blocksize_mask) || (size & blocksize_mask))  
1172                                 goto out;
1173                 }
1174         }
1175
1176         dio = kmalloc(sizeof(*dio), GFP_KERNEL);
1177         retval = -ENOMEM;
1178         if (!dio)
1179                 goto out;
1180
1181         /*
1182          * For block device access DIO_NO_LOCKING is used,
1183          *      neither readers nor writers do any locking at all
1184          * For regular files using DIO_LOCKING,
1185          *      readers need to grab i_mutex and i_alloc_sem
1186          *      writers need to grab i_alloc_sem only (i_mutex is already held)
1187          * For regular files using DIO_OWN_LOCKING,
1188          *      neither readers nor writers take any locks here
1189          */
1190         dio->lock_type = dio_lock_type;
1191         if (dio_lock_type != DIO_NO_LOCKING) {
1192                 /* watch out for a 0 len io from a tricksy fs */
1193                 if (rw == READ && end > offset) {
1194                         struct address_space *mapping;
1195
1196                         mapping = iocb->ki_filp->f_mapping;
1197                         if (dio_lock_type != DIO_OWN_LOCKING) {
1198                                 mutex_lock(&inode->i_mutex);
1199                                 release_i_mutex = 1;
1200                         }
1201
1202                         retval = filemap_write_and_wait_range(mapping, offset,
1203                                                               end - 1);
1204                         if (retval) {
1205                                 kfree(dio);
1206                                 goto out;
1207                         }
1208
1209                         if (dio_lock_type == DIO_OWN_LOCKING) {
1210                                 mutex_unlock(&inode->i_mutex);
1211                                 acquire_i_mutex = 1;
1212                         }
1213                 }
1214
1215                 if (dio_lock_type == DIO_LOCKING)
1216                         /* lockdep: not the owner will release it */
1217                         down_read_non_owner(&inode->i_alloc_sem);
1218         }
1219
1220         /*
1221          * For file extending writes updating i_size before data
1222          * writeouts complete can expose uninitialized blocks. So
1223          * even for AIO, we need to wait for i/o to complete before
1224          * returning in this case.
1225          */
1226         dio->is_async = !is_sync_kiocb(iocb) && !((rw & WRITE) &&
1227                 (end > i_size_read(inode)));
1228
1229         retval = direct_io_worker(rw, iocb, inode, iov, offset,
1230                                 nr_segs, blkbits, get_block, end_io, dio);
1231
1232         if (rw == READ && dio_lock_type == DIO_LOCKING)
1233                 release_i_mutex = 0;
1234
1235 out:
1236         if (release_i_mutex)
1237                 mutex_unlock(&inode->i_mutex);
1238         else if (acquire_i_mutex)
1239                 mutex_lock(&inode->i_mutex);
1240         return retval;
1241 }
1242 EXPORT_SYMBOL(__blockdev_direct_IO);