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;
60 spin_lock_irqsave(&dreq->dr_lock, flags);
61 bio->bi_private = dreq->dr_bios;
63 if (dreq->dr_error == 0)
64 dreq->dr_error = error;
65 spin_unlock_irqrestore(&dreq->dr_lock, flags);
67 if (atomic_dec_and_test(&dreq->dr_numreqs))
68 wake_up(&dreq->dr_wait);
73 static int can_be_merged(struct bio *bio, sector_t sector)
79 size = bio->bi_size >> 9;
80 return bio->bi_sector + size == sector ? 1 : 0;
84 int filter_alloc_iobuf(int rw, int num_pages, void **ret)
86 struct dio_request *dreq;
88 LASSERTF(rw == OBD_BRW_WRITE || rw == OBD_BRW_READ, "%x\n", rw);
90 OBD_ALLOC(dreq, sizeof(*dreq));
94 OBD_ALLOC(dreq->dr_pages, num_pages * sizeof(*dreq->dr_pages));
95 if (dreq->dr_pages == NULL)
98 OBD_ALLOC(dreq->dr_blocks,
99 MAX_BLOCKS_PER_PAGE * num_pages * sizeof(*dreq->dr_blocks));
100 if (dreq->dr_blocks == NULL)
103 dreq->dr_bios = NULL;
104 init_waitqueue_head(&dreq->dr_wait);
105 atomic_set(&dreq->dr_numreqs, 0);
106 spin_lock_init(&dreq->dr_lock);
107 dreq->dr_max_pages = num_pages;
114 OBD_FREE(dreq->dr_pages,
115 num_pages * sizeof(*dreq->dr_pages));
117 OBD_FREE(dreq, sizeof(*dreq));
122 void filter_free_iobuf(void *iobuf)
124 struct dio_request *dreq = iobuf;
125 int num_pages = dreq->dr_max_pages;
128 while (dreq->dr_bios) {
129 struct bio *bio = dreq->dr_bios;
130 dreq->dr_bios = bio->bi_private;
134 OBD_FREE(dreq->dr_blocks,
135 MAX_BLOCKS_PER_PAGE * num_pages * sizeof(*dreq->dr_blocks));
136 OBD_FREE(dreq->dr_pages,
137 num_pages * sizeof(*dreq->dr_pages));
138 OBD_FREE(dreq, sizeof(*dreq));
141 int filter_iobuf_add_page(struct obd_device *obd, void *iobuf,
142 struct inode *inode, struct page *page)
144 struct dio_request *dreq = iobuf;
146 LASSERT (dreq->dr_npages < dreq->dr_max_pages);
147 dreq->dr_pages[dreq->dr_npages++] = page;
152 int filter_do_bio(struct obd_device *obd, struct inode *inode,
153 struct dio_request *dreq, int rw)
155 int blocks_per_page = PAGE_SIZE >> inode->i_blkbits;
156 struct page **pages = dreq->dr_pages;
157 int npages = dreq->dr_npages;
158 unsigned long *blocks = dreq->dr_blocks;
159 int total_blocks = npages * blocks_per_page;
160 int sector_bits = inode->i_sb->s_blocksize_bits - 9;
161 unsigned int blocksize = inode->i_sb->s_blocksize;
162 struct bio *bio = NULL;
164 unsigned int page_offset;
173 LASSERT(dreq->dr_npages == npages);
174 LASSERT(total_blocks <= OBDFILTER_CREATED_SCRATCHPAD_ENTRIES);
176 for (page_idx = 0, block_idx = 0;
178 page_idx++, block_idx += blocks_per_page) {
180 page = pages[page_idx];
181 LASSERT (block_idx + blocks_per_page <= total_blocks);
183 for (i = 0, page_offset = 0;
185 i += nblocks, page_offset += blocksize * nblocks) {
189 if (blocks[block_idx + i] == 0) { /* hole */
190 LASSERT(rw == OBD_BRW_READ);
191 memset(kmap(page) + page_offset, 0, blocksize);
196 sector = blocks[block_idx + i] << sector_bits;
198 /* Additional contiguous file blocks? */
199 while (i + nblocks < blocks_per_page &&
200 (sector + nblocks*(blocksize>>9)) ==
201 (blocks[block_idx + i + nblocks] << sector_bits))
205 can_be_merged(bio, sector) &&
206 bio_add_page(bio, page,
207 blocksize * nblocks, page_offset) != 0)
208 continue; /* added this frag OK */
211 request_queue_t *q = bdev_get_queue(bio->bi_bdev);
213 /* Dang! I have to fragment this I/O */
214 CDEBUG(D_INODE, "bio++ sz %d vcnt %d(%d) "
215 "sectors %d(%d) psg %d(%d) hsg %d(%d)\n",
217 bio->bi_vcnt, bio->bi_max_vecs,
218 bio->bi_size >> 9, q->max_sectors,
219 bio_phys_segments(q, bio),
220 q->max_phys_segments,
221 bio_hw_segments(q, bio),
224 atomic_inc(&dreq->dr_numreqs);
225 rc = fsfilt_send_bio(rw, obd, inode, bio);
227 CERROR("Can't send bio: %d\n", rc);
228 /* OK do dec; we do the waiting */
229 atomic_dec(&dreq->dr_numreqs);
237 /* allocate new bio */
238 bio = bio_alloc(GFP_NOIO,
239 (npages - page_idx) * blocks_per_page);
241 CERROR ("Can't allocate bio\n");
246 bio->bi_bdev = inode->i_sb->s_bdev;
247 bio->bi_sector = sector;
248 bio->bi_end_io = dio_complete_routine;
249 bio->bi_private = dreq;
251 rc = bio_add_page(bio, page,
252 blocksize * nblocks, page_offset);
258 atomic_inc(&dreq->dr_numreqs);
259 rc = fsfilt_send_bio(rw, obd, inode, bio);
263 CERROR("Can't send bio: %d\n", rc);
264 /* OK do dec; we do the waiting */
265 atomic_dec(&dreq->dr_numreqs);
270 wait_event(dreq->dr_wait, atomic_read(&dreq->dr_numreqs) == 0);
277 static void filter_clear_page_cache(struct inode *inode, struct kiobuf *iobuf)
283 for (i = 0; i < iobuf->nr_pages ; i++) {
284 page = find_lock_page(inode->i_mapping,
285 iobuf->maplist[i]->index);
288 if (page->mapping != NULL) {
289 block_invalidatepage(page, 0);
290 truncate_complete_page(page);
293 page_cache_release(page);
298 /* Must be called with i_sem taken for writes; this will drop it */
299 int filter_direct_io(int rw, struct dentry *dchild, void *iobuf,
300 struct obd_export *exp, struct iattr *attr,
301 struct obd_trans_info *oti, void **wait_handle)
303 struct obd_device *obd = exp->exp_obd;
304 struct inode *inode = dchild->d_inode;
305 struct dio_request *dreq = iobuf;
309 LASSERTF(rw == OBD_BRW_WRITE || rw == OBD_BRW_READ, "%x\n", rw);
310 LASSERTF(dreq->dr_npages <= dreq->dr_max_pages, "%d,%d\n",
311 dreq->dr_npages, dreq->dr_max_pages);
313 if (dreq->dr_npages == 0)
316 if (dreq->dr_npages > OBDFILTER_CREATED_SCRATCHPAD_ENTRIES)
319 rc = fsfilt_map_inode_pages(obd, inode,
320 dreq->dr_pages, dreq->dr_npages,
322 obdfilter_created_scratchpad,
323 rw == OBD_BRW_WRITE, NULL);
325 if (rw == OBD_BRW_WRITE) {
328 filter_tally_write(&obd->u.filter,
334 if (attr->ia_size > inode->i_size)
335 attr->ia_valid |= ATTR_SIZE;
336 rc = fsfilt_setattr(obd, dchild,
337 oti->oti_handle, attr, 0);
342 rc2 = filter_finish_transno(exp, oti, 0);
344 CERROR("can't close transaction: %d\n", rc);
352 /* This is nearly osync_inode, without the waiting
353 rc = generic_osync_inode(inode, inode->i_mapping,
354 OSYNC_DATA|OSYNC_METADATA); */
355 rc = filemap_fdatawrite(inode->i_mapping);
356 rc2 = sync_mapping_buffers(inode->i_mapping);
359 rc2 = filemap_fdatawait(inode->i_mapping);
366 /* be careful to call this after fsync_inode_data_buffers has waited
367 * for IO to complete before we evict it from the cache */
368 filter_clear_page_cache(inode, iobuf);
370 RETURN(filter_do_bio(obd, inode, dreq, rw));
373 /* See if there are unallocated parts in given file region */
374 static int filter_range_is_mapped(struct inode *inode, obd_size offset, int len)
376 sector_t (*fs_bmap)(struct address_space *, sector_t) =
377 inode->i_mapping->a_ops->bmap;
380 /* We can't know if we are overwriting or not */
384 offset >>= inode->i_blkbits;
385 len >>= inode->i_blkbits;
387 for (j = 0; j <= len; j++)
388 if (fs_bmap(inode->i_mapping, offset + j) == 0)
394 int filter_commitrw_write(struct obd_export *exp, struct obdo *oa,
395 int objcount, struct obd_ioobj *obj, int niocount,
396 struct niobuf_local *res, struct obd_trans_info *oti,
399 struct niobuf_local *lnb;
400 struct dio_request *dreq = NULL;
401 struct lvfs_run_ctxt saved;
402 struct fsfilt_objinfo fso;
403 struct iattr iattr = { 0 };
404 struct inode *inode = NULL;
405 unsigned long now = jiffies;
406 int i, err, cleanup_phase = 0;
407 struct obd_device *obd = exp->exp_obd;
411 LASSERT(oti != NULL);
412 LASSERT(objcount == 1);
413 LASSERT(current->journal_info == NULL);
418 rc = filter_alloc_iobuf(OBD_BRW_WRITE, obj->ioo_bufcnt, (void **)&dreq);
423 fso.fso_dentry = res->dentry;
424 fso.fso_bufcnt = obj->ioo_bufcnt;
425 inode = res->dentry->d_inode;
427 for (i = 0, lnb = res; i < obj->ioo_bufcnt; i++, lnb++) {
430 /* If overwriting an existing block, we don't need a grant */
431 if (!(lnb->flags & OBD_BRW_GRANTED) && lnb->rc == -ENOSPC &&
432 filter_range_is_mapped(inode, lnb->offset, lnb->len))
435 if (lnb->rc) { /* ENOSPC, network RPC error, etc. */
436 CDEBUG(D_INODE, "Skipping [%d] == %d\n", i, lnb->rc);
440 err = filter_iobuf_add_page(obd, dreq, inode, lnb->page);
443 total_size += lnb->len;
445 /* we expect these pages to be in offset order, but we'll
447 this_size = lnb->offset + lnb->len;
448 if (this_size > iattr.ia_size)
449 iattr.ia_size = this_size;
452 /* I use this when I'm checking our lovely 1M I/Os reach the disk -eeb */
453 if (total_size != (1<<20))
454 CWARN("total size %d (%d pages)\n",
455 total_size, total_size/PAGE_SIZE);
457 push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
461 oti->oti_handle = fsfilt_brw_start(obd, objcount, &fso, niocount, res,
463 if (IS_ERR(oti->oti_handle)) {
465 rc = PTR_ERR(oti->oti_handle);
466 CDEBUG(rc == -ENOSPC ? D_INODE : D_ERROR,
467 "error starting transaction: rc = %d\n", rc);
468 oti->oti_handle = NULL;
471 /* have to call fsfilt_commit() from this point on */
473 fsfilt_check_slow(now, obd_timeout, "brw_start");
475 iattr_from_obdo(&iattr,oa,OBD_MD_FLATIME|OBD_MD_FLMTIME|OBD_MD_FLCTIME);
476 /* filter_direct_io drops i_sem */
477 rc = filter_direct_io(OBD_BRW_WRITE, res->dentry, dreq, exp, &iattr,
480 obdo_from_inode(oa, inode, FILTER_VALID_FLAGS);
482 fsfilt_check_slow(now, obd_timeout, "direct_io");
484 err = fsfilt_commit(obd, obd->u.filter.fo_sb, inode, oti->oti_handle,
489 if (obd_sync_filter && !err)
490 LASSERT(oti->oti_transno <= obd->obd_last_committed);
492 fsfilt_check_slow(now, obd_timeout, "commitrw commit");
495 filter_grant_commit(exp, niocount, res);
497 switch (cleanup_phase) {
499 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
500 LASSERT(current->journal_info == NULL);
502 filter_free_iobuf(dreq);
504 filter_free_dio_pages(objcount, obj, niocount, res);