1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
4 * linux/fs/obdfilter/filter_io.c
6 * Copyright (c) 2001-2003 Cluster File Systems, Inc.
7 * Author: Peter Braam <braam@clusterfs.com>
8 * Author: Andreas Dilger <adilger@clusterfs.com>
9 * Author: Phil Schwan <phil@clusterfs.com>
11 * This file is part of Lustre, http://www.lustre.org.
13 * Lustre is free software; you can redistribute it and/or
14 * modify it under the terms of version 2 of the GNU General Public
15 * License as published by the Free Software Foundation.
17 * Lustre is distributed in the hope that it will be useful,
18 * but WITHOUT ANY WARRANTY; without even the implied warranty of
19 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
20 * GNU General Public License for more details.
22 * You should have received a copy of the GNU General Public License
23 * along with Lustre; if not, write to the Free Software
24 * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
27 #include <linux/config.h>
28 #include <linux/module.h>
29 #include <linux/pagemap.h> // XXX kill me soon
30 #include <linux/version.h>
31 #include <linux/buffer_head.h>
33 #define DEBUG_SUBSYSTEM S_FILTER
35 #include <linux/obd_class.h>
36 #include <linux/lustre_fsfilt.h>
37 #include "filter_internal.h"
39 #warning "implement writeback mode -bzzz"
41 /* 512byte block min */
42 #define MAX_BLOCKS_PER_PAGE (PAGE_SIZE / 512)
44 atomic_t dr_numreqs; /* number of reqs being processed */
45 struct bio *dr_bios; /* list of completed bios */
46 wait_queue_head_t dr_wait;
50 struct page **dr_pages;
51 unsigned long *dr_blocks;
55 static int dio_complete_routine(struct bio *bio, unsigned int done, int error)
57 struct dio_request *dreq = bio->bi_private;
61 CWARN("gets called against non-complete bio 0x%p: %d/%d/%d\n",
62 bio, bio->bi_size, done, error);
67 CERROR("***** bio->bi_private is NULL! This should never "
68 "happen. Normally, I would crash here, but instead I "
69 "will dump the bio contents to the console. Please "
70 "report this to CFS, along with any interesting messages "
71 "leading up to this point (like SCSI errors, perhaps). "
72 "Because bi_private is NULL, I can't wake up the thread "
73 "that initiated this I/O -- so you will probably have to "
75 CERROR("bi_next: %p, bi_flags: %lx, bi_rw: %lu, bi_vcnt: %d, "
76 "bi_idx: %d, bi->size: %d, bi_end_io: %p, bi_cnt: %d, "
77 "bi_private: %p\n", bio->bi_next, bio->bi_flags,
78 bio->bi_rw, bio->bi_vcnt, bio->bi_idx, bio->bi_size,
79 bio->bi_end_io, atomic_read(&bio->bi_cnt),
84 spin_lock_irqsave(&dreq->dr_lock, flags);
85 bio->bi_private = dreq->dr_bios;
87 if (dreq->dr_error == 0)
88 dreq->dr_error = error;
89 spin_unlock_irqrestore(&dreq->dr_lock, flags);
91 if (atomic_dec_and_test(&dreq->dr_numreqs))
92 wake_up(&dreq->dr_wait);
97 static int can_be_merged(struct bio *bio, sector_t sector)
103 size = bio->bi_size >> 9;
104 return bio->bi_sector + size == sector ? 1 : 0;
108 int filter_alloc_iobuf(int rw, int num_pages, void **ret)
110 struct dio_request *dreq;
112 LASSERTF(rw == OBD_BRW_WRITE || rw == OBD_BRW_READ, "%x\n", rw);
114 OBD_ALLOC(dreq, sizeof(*dreq));
118 OBD_ALLOC(dreq->dr_pages, num_pages * sizeof(*dreq->dr_pages));
119 if (dreq->dr_pages == NULL)
122 OBD_ALLOC(dreq->dr_blocks,
123 MAX_BLOCKS_PER_PAGE * num_pages * sizeof(*dreq->dr_blocks));
124 if (dreq->dr_blocks == NULL)
127 dreq->dr_bios = NULL;
128 init_waitqueue_head(&dreq->dr_wait);
129 atomic_set(&dreq->dr_numreqs, 0);
130 spin_lock_init(&dreq->dr_lock);
131 dreq->dr_max_pages = num_pages;
138 OBD_FREE(dreq->dr_pages,
139 num_pages * sizeof(*dreq->dr_pages));
141 OBD_FREE(dreq, sizeof(*dreq));
146 void filter_free_iobuf(void *iobuf)
148 struct dio_request *dreq = iobuf;
149 int num_pages = dreq->dr_max_pages;
152 while (dreq->dr_bios) {
153 struct bio *bio = dreq->dr_bios;
154 dreq->dr_bios = bio->bi_private;
158 OBD_FREE(dreq->dr_blocks,
159 MAX_BLOCKS_PER_PAGE * num_pages * sizeof(*dreq->dr_blocks));
160 OBD_FREE(dreq->dr_pages,
161 num_pages * sizeof(*dreq->dr_pages));
162 OBD_FREE(dreq, sizeof(*dreq));
165 int filter_iobuf_add_page(struct obd_device *obd, void *iobuf,
166 struct inode *inode, struct page *page)
168 struct dio_request *dreq = iobuf;
170 LASSERT (dreq->dr_npages < dreq->dr_max_pages);
171 dreq->dr_pages[dreq->dr_npages++] = page;
176 int filter_do_bio(struct obd_device *obd, struct inode *inode,
177 struct dio_request *dreq, int rw)
179 int blocks_per_page = PAGE_SIZE >> inode->i_blkbits;
180 struct page **pages = dreq->dr_pages;
181 int npages = dreq->dr_npages;
182 unsigned long *blocks = dreq->dr_blocks;
183 int total_blocks = npages * blocks_per_page;
184 int sector_bits = inode->i_sb->s_blocksize_bits - 9;
185 unsigned int blocksize = inode->i_sb->s_blocksize;
186 struct bio *bio = NULL;
188 unsigned int page_offset;
197 LASSERT(dreq->dr_npages == npages);
198 LASSERT(total_blocks <= OBDFILTER_CREATED_SCRATCHPAD_ENTRIES);
200 for (page_idx = 0, block_idx = 0;
202 page_idx++, block_idx += blocks_per_page) {
204 page = pages[page_idx];
205 LASSERT (block_idx + blocks_per_page <= total_blocks);
207 for (i = 0, page_offset = 0;
209 i += nblocks, page_offset += blocksize * nblocks) {
213 if (blocks[block_idx + i] == 0) { /* hole */
214 LASSERT(rw == OBD_BRW_READ);
215 memset(kmap(page) + page_offset, 0, blocksize);
220 sector = blocks[block_idx + i] << sector_bits;
222 /* Additional contiguous file blocks? */
223 while (i + nblocks < blocks_per_page &&
224 (sector + nblocks*(blocksize>>9)) ==
225 (blocks[block_idx + i + nblocks] << sector_bits))
229 can_be_merged(bio, sector) &&
230 bio_add_page(bio, page,
231 blocksize * nblocks, page_offset) != 0)
232 continue; /* added this frag OK */
235 request_queue_t *q = bdev_get_queue(bio->bi_bdev);
237 /* Dang! I have to fragment this I/O */
238 CDEBUG(D_INODE, "bio++ sz %d vcnt %d(%d) "
239 "sectors %d(%d) psg %d(%d) hsg %d(%d)\n",
241 bio->bi_vcnt, bio->bi_max_vecs,
242 bio->bi_size >> 9, q->max_sectors,
243 bio_phys_segments(q, bio),
244 q->max_phys_segments,
245 bio_hw_segments(q, bio),
248 atomic_inc(&dreq->dr_numreqs);
249 rc = fsfilt_send_bio(rw, obd, inode, bio);
251 CERROR("Can't send bio: %d\n", rc);
252 /* OK do dec; we do the waiting */
253 atomic_dec(&dreq->dr_numreqs);
261 /* allocate new bio */
262 bio = bio_alloc(GFP_NOIO,
263 (npages - page_idx) * blocks_per_page);
265 CERROR ("Can't allocate bio\n");
270 bio->bi_bdev = inode->i_sb->s_bdev;
271 bio->bi_sector = sector;
272 bio->bi_end_io = dio_complete_routine;
273 bio->bi_private = dreq;
275 rc = bio_add_page(bio, page,
276 blocksize * nblocks, page_offset);
282 atomic_inc(&dreq->dr_numreqs);
283 rc = fsfilt_send_bio(rw, obd, inode, bio);
287 CERROR("Can't send bio: %d\n", rc);
288 /* OK do dec; we do the waiting */
289 atomic_dec(&dreq->dr_numreqs);
294 wait_event(dreq->dr_wait, atomic_read(&dreq->dr_numreqs) == 0);
301 /* These are our hacks to keep our directio/bh IO coherent with ext3's
302 * page cache use. Most notably ext3 reads file data into the page
303 * cache when it is zeroing the tail of partial-block truncates and
304 * leaves it there, sometimes generating io from it at later truncates.
305 * This removes the partial page and its buffers from the page cache,
306 * so it should only ever cause a wait in rare cases, as otherwise we
307 * always do full-page IO to the OST.
309 * The call to truncate_complete_page() will call journal_invalidatepage()
310 * to free the buffers and drop the page from cache. The buffers should
311 * not be dirty, because we already called fdatasync/fdatawait on them.
313 static int filter_clear_page_cache(struct inode *inode,
314 struct dio_request *iobuf)
319 /* This is nearly generic_osync_inode, without the waiting on the inode
320 rc = generic_osync_inode(inode, inode->i_mapping,
321 OSYNC_DATA|OSYNC_METADATA);
323 rc = filemap_fdatawrite(inode->i_mapping);
324 rc2 = sync_mapping_buffers(inode->i_mapping);
327 rc2 = filemap_fdatawait(inode->i_mapping);
333 /* be careful to call this after fsync_inode_data_buffers has waited
334 * for IO to complete before we evict it from the cache */
335 for (i = 0; i < iobuf->dr_npages; i++) {
336 page = find_lock_page(inode->i_mapping,
337 iobuf->dr_pages[i]->index);
340 if (page->mapping != NULL) {
341 wait_on_page_writeback(page);
342 ll_truncate_complete_page(page);
346 page_cache_release(page);
350 /* Must be called with i_sem taken for writes; this will drop it */
351 int filter_direct_io(int rw, struct dentry *dchild, void *iobuf,
352 struct obd_export *exp, struct iattr *attr,
353 struct obd_trans_info *oti, void **wait_handle)
355 struct obd_device *obd = exp->exp_obd;
356 struct inode *inode = dchild->d_inode;
357 struct dio_request *dreq = iobuf;
361 LASSERTF(rw == OBD_BRW_WRITE || rw == OBD_BRW_READ, "%x\n", rw);
362 LASSERTF(dreq->dr_npages <= dreq->dr_max_pages, "%d,%d\n",
363 dreq->dr_npages, dreq->dr_max_pages);
365 if (dreq->dr_npages == 0)
368 if (dreq->dr_npages > OBDFILTER_CREATED_SCRATCHPAD_ENTRIES)
371 rc = fsfilt_map_inode_pages(obd, inode,
372 dreq->dr_pages, dreq->dr_npages,
374 obdfilter_created_scratchpad,
375 rw == OBD_BRW_WRITE, NULL);
377 if (rw == OBD_BRW_WRITE) {
380 filter_tally_write(&obd->u.filter,
386 if (attr->ia_size > inode->i_size)
387 attr->ia_valid |= ATTR_SIZE;
388 rc = fsfilt_setattr(obd, dchild,
389 oti->oti_handle, attr, 0);
394 rc2 = filter_finish_transno(exp, oti, 0);
396 CERROR("can't close transaction: %d\n", rc);
404 rc = filter_clear_page_cache(inode, dreq);
408 RETURN(filter_do_bio(obd, inode, dreq, rw));
411 /* See if there are unallocated parts in given file region */
412 static int filter_range_is_mapped(struct inode *inode, obd_size offset, int len)
414 sector_t (*fs_bmap)(struct address_space *, sector_t) =
415 inode->i_mapping->a_ops->bmap;
418 /* We can't know if we are overwriting or not */
422 offset >>= inode->i_blkbits;
423 len >>= inode->i_blkbits;
425 for (j = 0; j <= len; j++)
426 if (fs_bmap(inode->i_mapping, offset + j) == 0)
432 int filter_commitrw_write(struct obd_export *exp, struct obdo *oa,
433 int objcount, struct obd_ioobj *obj, int niocount,
434 struct niobuf_local *res, struct obd_trans_info *oti,
437 struct niobuf_local *lnb;
438 struct dio_request *dreq = NULL;
439 struct lvfs_run_ctxt saved;
440 struct fsfilt_objinfo fso;
441 struct iattr iattr = { 0 };
442 struct inode *inode = NULL;
443 unsigned long now = jiffies;
444 int i, err, cleanup_phase = 0;
445 struct obd_device *obd = exp->exp_obd;
449 LASSERT(oti != NULL);
450 LASSERT(objcount == 1);
451 LASSERT(current->journal_info == NULL);
456 rc = filter_alloc_iobuf(OBD_BRW_WRITE, obj->ioo_bufcnt, (void **)&dreq);
461 fso.fso_dentry = res->dentry;
462 fso.fso_bufcnt = obj->ioo_bufcnt;
463 inode = res->dentry->d_inode;
465 for (i = 0, lnb = res; i < obj->ioo_bufcnt; i++, lnb++) {
468 /* If overwriting an existing block, we don't need a grant */
469 if (!(lnb->flags & OBD_BRW_GRANTED) && lnb->rc == -ENOSPC &&
470 filter_range_is_mapped(inode, lnb->offset, lnb->len))
473 if (lnb->rc) { /* ENOSPC, network RPC error, etc. */
474 CDEBUG(D_INODE, "Skipping [%d] == %d\n", i, lnb->rc);
478 err = filter_iobuf_add_page(obd, dreq, inode, lnb->page);
481 total_size += lnb->len;
483 /* we expect these pages to be in offset order, but we'll
485 this_size = lnb->offset + lnb->len;
486 if (this_size > iattr.ia_size)
487 iattr.ia_size = this_size;
490 /* I use this when I'm checking our lovely 1M I/Os reach the disk -eeb */
491 if (total_size != (1<<20))
492 CWARN("total size %d (%d pages)\n",
493 total_size, total_size/PAGE_SIZE);
495 push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
499 oti->oti_handle = fsfilt_brw_start(obd, objcount, &fso, niocount, res,
501 if (IS_ERR(oti->oti_handle)) {
503 rc = PTR_ERR(oti->oti_handle);
504 CDEBUG(rc == -ENOSPC ? D_INODE : D_ERROR,
505 "error starting transaction: rc = %d\n", rc);
506 oti->oti_handle = NULL;
509 /* have to call fsfilt_commit() from this point on */
511 fsfilt_check_slow(now, obd_timeout, "brw_start");
513 iattr_from_obdo(&iattr,oa,OBD_MD_FLATIME|OBD_MD_FLMTIME|OBD_MD_FLCTIME);
514 /* filter_direct_io drops i_sem */
515 rc = filter_direct_io(OBD_BRW_WRITE, res->dentry, dreq, exp, &iattr,
518 obdo_from_inode(oa, inode, FILTER_VALID_FLAGS);
520 fsfilt_check_slow(now, obd_timeout, "direct_io");
522 err = fsfilt_commit(obd, obd->u.filter.fo_sb, inode, oti->oti_handle,
527 if (obd_sync_filter && !err)
528 LASSERTF(oti->oti_transno <= obd->obd_last_committed,
529 "oti_transno "LPU64" last_committed "LPU64"\n",
530 oti->oti_transno, obd->obd_last_committed);
532 fsfilt_check_slow(now, obd_timeout, "commitrw commit");
535 filter_grant_commit(exp, niocount, res);
537 switch (cleanup_phase) {
539 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
540 LASSERT(current->journal_info == NULL);
542 filter_free_iobuf(dreq);
544 filter_free_dio_pages(objcount, obj, niocount, res);