lustre/obdfilter/filter_io_26.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  linux/fs/obdfilter/filter_io.c
   5  *
   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>
  10  *
  11  *   This file is part of Lustre, http://www.lustre.org.
  12  *
  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.
  16  *
  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.
  21  *
  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.
  25  */
  26
  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>
  32
  33 #define DEBUG_SUBSYSTEM S_FILTER
  34
  35 #include <linux/obd_class.h>
  36 #include <linux/lustre_fsfilt.h>
  37 #include "filter_internal.h"
  38
  39 #warning "implement writeback mode -bzzz"
  40
  41 /* 512byte block min */
  42 #define MAX_BLOCKS_PER_PAGE (PAGE_SIZE / 512)
  43 struct dio_request {
  44         atomic_t numreqs;       /* number of reqs being processed */
  45         struct bio *bio_current;/* bio currently being constructed */
  46         struct bio *bio_list;   /* list of completed bios */
  47         wait_queue_head_t dr_wait;
  48         int dr_num_pages;
  49         int dr_rw;
  50         int dr_error;
  51         int dr_created[MAX_BLOCKS_PER_PAGE];
  52         unsigned long dr_blocks[MAX_BLOCKS_PER_PAGE];
  53         spinlock_t dr_lock;
  54
  55 };
  56
  57 static int dio_complete_routine(struct bio *bio, unsigned int done, int error)
  58 {
  59         struct dio_request *dreq = bio->bi_private;
  60         unsigned long flags;
  61
  62         spin_lock_irqsave(&dreq->dr_lock, flags);
  63         bio->bi_private = dreq->bio_list;
  64         dreq->bio_list = bio;
  65         spin_unlock_irqrestore(&dreq->dr_lock, flags);
  66         if (atomic_dec_and_test(&dreq->numreqs))
  67                 wake_up(&dreq->dr_wait);
  68         if (dreq->dr_error == 0)
  69                 dreq->dr_error = error;
  70         return 0;
  71 }
  72
  73 static int can_be_merged(struct bio *bio, sector_t sector)
  74 {
  75         unsigned int size;
  76         if (!bio)
  77                 return 0;
  78
  79         size = bio->bi_size >> 9;
  80         return bio->bi_sector + size == sector ? 1 : 0;
  81 }
  82 int filter_alloc_iobuf(int rw, int num_pages, void **ret)
  83 {
  84         struct dio_request *dreq;
  85
  86         LASSERTF(rw == OBD_BRW_WRITE || rw == OBD_BRW_READ, "%x\n", rw);
  87
  88         OBD_ALLOC(dreq, sizeof(*dreq));
  89         if (dreq == NULL)
  90                 RETURN(-ENOMEM);
  91
  92         dreq->bio_list = NULL;
  93         init_waitqueue_head(&dreq->dr_wait);
  94         atomic_set(&dreq->numreqs, 0);
  95         spin_lock_init(&dreq->dr_lock);
  96         dreq->dr_num_pages = num_pages;
  97         dreq->dr_rw = rw;
  98
  99         *ret = dreq;
 100         RETURN(0);
 101 }
 102
 103 void filter_free_iobuf(void *iobuf)
 104 {
 105         struct dio_request *dreq = iobuf;
 106
 107         /* free all bios */
 108         while (dreq->bio_list) {
 109                 struct bio *bio = dreq->bio_list;
 110                 dreq->bio_list = bio->bi_private;
 111                 bio_put(bio);
 112         }
 113
 114         OBD_FREE(dreq, sizeof(*dreq));
 115 }
 116
 117 int filter_iobuf_add_page(struct obd_device *obd, void *iobuf,
 118                           struct inode *inode, struct page *page)
 119 {
 120         struct dio_request *dreq = iobuf;
 121         int blocks_per_page = PAGE_SIZE >> inode->i_blkbits;
 122         unsigned int len = inode->i_sb->s_blocksize, offs;
 123         struct bio *bio = dreq->bio_current;
 124         sector_t sector;
 125         int k, rc;
 126         ENTRY;
 127
 128         /* get block number for next page */
 129         rc = fsfilt_map_inode_pages(obd, inode, &page, 1, dreq->dr_blocks,
 130                                     dreq->dr_created,
 131                                     dreq->dr_rw == OBD_BRW_WRITE, NULL);
 132         if (rc)
 133                 RETURN(rc);
 134
 135         for (k = 0, offs = 0; k < blocks_per_page; k++, offs += len) {
 136                 if (dreq->dr_created[k] == -1) {
 137                         memset(kmap(page) + offs, 0, len);
 138                         kunmap(page);
 139                         continue;
 140                 }
 141
 142                 sector = dreq->dr_blocks[k] <<(inode->i_sb->s_blocksize_bits-9);
 143
 144                 if (!bio || !can_be_merged(bio, sector) ||
 145                     !bio_add_page(bio, page, len, offs)) {
 146                         if (bio) {
 147                                 atomic_inc(&dreq->numreqs);
 148                                 /* FIXME
 149                                 filter_tally_write(&obd->u.filter,dreq->maplist,
 150                                                    dreq->nr_pages,dreq->blocks,
 151                                                    blocks_per_page);
 152                                 */
 153                                 fsfilt_send_bio(dreq->dr_rw, obd, inode, bio);
 154                                 dreq->bio_current = bio = NULL;
 155                         }
 156                         /* allocate new bio */
 157                         dreq->bio_current = bio =
 158                                 bio_alloc(GFP_NOIO, dreq->dr_num_pages *
 159                                                     blocks_per_page);
 160                         bio->bi_bdev = inode->i_sb->s_bdev;
 161                         bio->bi_sector = sector;
 162                         bio->bi_end_io = dio_complete_routine;
 163                         bio->bi_private = dreq;
 164
 165                         if (!bio_add_page(bio, page, len, offs))
 166                                 LBUG();
 167                 }
 168         }
 169         dreq->dr_num_pages--;
 170
 171         RETURN(0);
 172 }
 173
 174 static void filter_clear_page_cache(struct inode *inode, struct kiobuf *iobuf)
 175 {
 176 #if 0
 177         struct page *page;
 178         int i;
 179
 180         for (i = 0; i < iobuf->nr_pages ; i++) {
 181                 page = find_lock_page(inode->i_mapping,
 182                                       iobuf->maplist[i]->index);
 183                 if (page == NULL)
 184                         continue;
 185                 if (page->mapping != NULL) {
 186                         block_invalidatepage(page, 0);
 187                         truncate_complete_page(page);
 188                 }
 189                 unlock_page(page);
 190                 page_cache_release(page);
 191         }
 192 #endif
 193 }
 194
 195 /* Must be called with i_sem taken for writes; this will drop it */
 196 int filter_direct_io(int rw, struct dentry *dchild, void *iobuf,
 197                      struct obd_export *exp, struct iattr *attr,
 198                      struct obd_trans_info *oti, void **wait_handle)
 199 {
 200         struct dio_request *dreq = iobuf;
 201         struct inode *inode = dchild->d_inode;
 202         int rc;
 203         ENTRY;
 204
 205         LASSERTF(rw == OBD_BRW_WRITE || rw == OBD_BRW_READ, "%x\n", rw);
 206
 207         /* This is nearly osync_inode, without the waiting
 208         rc = generic_osync_inode(inode, inode->i_mapping,
 209                                  OSYNC_DATA|OSYNC_METADATA); */
 210         rc = filemap_fdatawrite(inode->i_mapping);
 211         if (rc == 0)
 212                 rc = sync_mapping_buffers(inode->i_mapping);
 213         if (rc == 0)
 214                 rc = filemap_fdatawait(inode->i_mapping);
 215         if (rc < 0)
 216                 GOTO(cleanup, rc);
 217
 218         if (rw == OBD_BRW_WRITE)
 219                 up(&inode->i_sem);
 220
 221         /* be careful to call this after fsync_inode_data_buffers has waited
 222          * for IO to complete before we evict it from the cache */
 223         filter_clear_page_cache(inode, iobuf);
 224
 225         if (dreq->bio_current != NULL) {
 226                 atomic_inc(&dreq->numreqs);
 227                 fsfilt_send_bio(rw, exp->exp_obd, inode, dreq->bio_current);
 228                 dreq->bio_current = NULL;
 229         }
 230
 231         /* time to wait for I/O completion */
 232         wait_event(dreq->dr_wait, atomic_read(&dreq->numreqs) == 0);
 233
 234         rc = dreq->dr_error;
 235         if (rw == OBD_BRW_WRITE && rc == 0) {
 236                 /* FIXME:
 237                 filter_tally_write(&obd->u.filter, dreq->maplist,
 238                                    dreq->nr_pages, dreq->blocks,
 239                                    blocks_per_page);
 240                 */
 241
 242                 if (attr->ia_size > inode->i_size) {
 243                         CDEBUG(D_INFO, "setting i_size to "LPU64"\n",
 244                                attr->ia_size);
 245
 246                         attr->ia_valid |= ATTR_SIZE;
 247                         down(&inode->i_sem);
 248                         fsfilt_setattr(exp->exp_obd, dchild, oti->oti_handle,
 249                                        attr, 0);
 250                         up(&inode->i_sem);
 251                 }
 252         }
 253
 254 cleanup:
 255         RETURN(rc);
 256 }
 257
 258 /* See if there are unallocated parts in given file region */
 259 static int filter_range_is_mapped(struct inode *inode, obd_size offset, int len)
 260 {
 261         sector_t (*fs_bmap)(struct address_space *, sector_t) =
 262                 inode->i_mapping->a_ops->bmap;
 263         int j;
 264
 265         /* We can't know if we are overwriting or not */
 266         if (fs_bmap == NULL)
 267                 return 0;
 268
 269         offset >>= inode->i_blkbits;
 270         len >>= inode->i_blkbits;
 271
 272         for (j = 0; j <= len; j++)
 273                 if (fs_bmap(inode->i_mapping, offset + j) == 0)
 274                         return 0;
 275
 276         return 1;
 277 }
 278
 279 int filter_commitrw_write(struct obd_export *exp, struct obdo *oa,
 280                           int objcount, struct obd_ioobj *obj, int niocount,
 281                           struct niobuf_local *res, struct obd_trans_info *oti,
 282                           int rc)
 283 {
 284         struct niobuf_local *lnb;
 285         struct dio_request *dreq = NULL;
 286         struct lvfs_run_ctxt saved;
 287         struct fsfilt_objinfo fso;
 288         struct iattr iattr = { 0 };
 289         struct inode *inode = NULL;
 290         unsigned long now = jiffies;
 291         int i, err, cleanup_phase = 0;
 292         struct obd_device *obd = exp->exp_obd;
 293
 294         ENTRY;
 295
 296         LASSERT(oti != NULL);
 297         LASSERT(objcount == 1);
 298         LASSERT(current->journal_info == NULL);
 299
 300         if (rc != 0)
 301                 GOTO(cleanup, rc);
 302
 303         inode = res->dentry->d_inode;
 304
 305         rc = filter_alloc_iobuf(OBD_BRW_WRITE, obj->ioo_bufcnt, (void **)&dreq);
 306         if (rc)
 307                 GOTO(cleanup, rc);
 308
 309         cleanup_phase = 1;
 310         fso.fso_dentry = res->dentry;
 311         fso.fso_bufcnt = obj->ioo_bufcnt;
 312
 313         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 314         cleanup_phase = 2;
 315
 316         generic_osync_inode(inode, inode->i_mapping, OSYNC_DATA|OSYNC_METADATA);
 317
 318         oti->oti_handle = fsfilt_brw_start(obd, objcount, &fso, niocount, res,
 319                                            oti);
 320         if (IS_ERR(oti->oti_handle)) {
 321                 rc = PTR_ERR(oti->oti_handle);
 322                 CDEBUG(rc == -ENOSPC ? D_INODE : D_ERROR,
 323                        "error starting transaction: rc = %d\n", rc);
 324                 oti->oti_handle = NULL;
 325                 GOTO(cleanup, rc);
 326         }
 327
 328         /* have to call fsfilt_commit() from this point on */
 329
 330         if (time_after(jiffies, now + 15 * HZ))
 331                 CERROR("slow brw_start %lus\n", (jiffies - now) / HZ);
 332
 333         down(&inode->i_sem);
 334         for (i = 0, lnb = res; i < obj->ioo_bufcnt; i++, lnb++) {
 335                 loff_t this_size;
 336
 337                 /* If overwriting an existing block, we don't need a grant */
 338                 if (!(lnb->flags & OBD_BRW_GRANTED) && lnb->rc == -ENOSPC &&
 339                     filter_range_is_mapped(inode, lnb->offset, lnb->len))
 340                         lnb->rc = 0;
 341
 342                 if (lnb->rc) /* ENOSPC, network RPC error, etc. */
 343                         continue;
 344
 345                 err = filter_iobuf_add_page(obd, dreq, inode, lnb->page);
 346                 if (err != 0) {
 347                         lnb->rc = err;
 348                         continue;
 349                 }
 350
 351                 /* we expect these pages to be in offset order, but we'll
 352                  * be forgiving */
 353                 this_size = lnb->offset + lnb->len;
 354                 if (this_size > iattr.ia_size)
 355                         iattr.ia_size = this_size;
 356         }
 357
 358         iattr_from_obdo(&iattr,oa,OBD_MD_FLATIME|OBD_MD_FLMTIME|OBD_MD_FLCTIME);
 359         rc = filter_direct_io(OBD_BRW_WRITE, res->dentry, dreq, exp, &iattr,
 360                               oti, NULL);
 361         rc = filter_finish_transno(exp, oti, rc);
 362
 363         if (time_after(jiffies, now + 15 * HZ))
 364                 CERROR("slow direct_io %lus\n", (jiffies - now) / HZ);
 365
 366
 367         err = fsfilt_commit(obd, obd->u.filter.fo_sb, inode, oti->oti_handle,
 368                             obd_sync_filter);
 369         if (err)
 370                 rc = err;
 371
 372         if (obd_sync_filter)
 373                 LASSERT(oti->oti_transno <= obd->obd_last_committed);
 374
 375         if (time_after(jiffies, now + 15 * HZ))
 376                 CERROR("slow commitrw commit %lus\n", (jiffies - now) / HZ);
 377
 378 cleanup:
 379         filter_grant_commit(exp, niocount, res);
 380
 381         switch (cleanup_phase) {
 382         case 2:
 383                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 384                 LASSERT(current->journal_info == NULL);
 385         case 1:
 386                 filter_free_iobuf(dreq);
 387         case 0:
 388                 filter_free_dio_pages(objcount, obj, niocount, res);
 389                 f_dput(res->dentry);
 390         }
 391
 392         RETURN(rc);
 393 }