lustre/obdfilter/filter_io.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 #define DEBUG_SUBSYSTEM S_FILTER
  28
  29 #include <linux/config.h>
  30 #include <linux/module.h>
  31 #include <linux/pagemap.h> // XXX kill me soon
  32 #include <linux/version.h>
  33
  34 #include <linux/obd_class.h>
  35 #include <linux/lustre_fsfilt.h>
  36 #include "filter_internal.h"
  37
  38 static int filter_start_page_read(struct inode *inode, struct niobuf_local *lnb)
  39 {
  40         struct address_space *mapping = inode->i_mapping;
  41         struct page *page;
  42         unsigned long index = lnb->offset >> PAGE_SHIFT;
  43         int rc;
  44
  45         page = grab_cache_page(mapping, index); /* locked page */
  46         if (IS_ERR(page))
  47                 return lnb->rc = PTR_ERR(page);
  48
  49         LASSERT(page->mapping == mapping);
  50
  51         lnb->page = page;
  52
  53         if (inode->i_size < lnb->offset + lnb->len - 1)
  54                 lnb->rc = inode->i_size - lnb->offset;
  55         else
  56                 lnb->rc = lnb->len;
  57
  58         if (PageUptodate(page)) {
  59                 unlock_page(page);
  60                 return 0;
  61         }
  62
  63         rc = mapping->a_ops->readpage(NULL, page);
  64         if (rc < 0) {
  65                 CERROR("page index %lu, rc = %d\n", index, rc);
  66                 lnb->page = NULL;
  67                 page_cache_release(page);
  68                 return lnb->rc = rc;
  69         }
  70
  71         return 0;
  72 }
  73
  74 static int filter_finish_page_read(struct niobuf_local *lnb)
  75 {
  76         if (lnb->page == NULL)
  77                 return 0;
  78
  79         if (PageUptodate(lnb->page))
  80                 return 0;
  81
  82         wait_on_page(lnb->page);
  83         if (!PageUptodate(lnb->page)) {
  84                 CERROR("page index %lu/offset "LPX64" not uptodate\n",
  85                        lnb->page->index, lnb->offset);
  86                 GOTO(err_page, lnb->rc = -EIO);
  87         }
  88         if (PageError(lnb->page)) {
  89                 CERROR("page index %lu/offset "LPX64" has error\n",
  90                        lnb->page->index, lnb->offset);
  91                 GOTO(err_page, lnb->rc = -EIO);
  92         }
  93
  94         return 0;
  95
  96 err_page:
  97         page_cache_release(lnb->page);
  98         lnb->page = NULL;
  99         return lnb->rc;
 100 }
 101
 102 static struct page *lustre_get_page_write(struct inode *inode,
 103                                           unsigned long index)
 104 {
 105         struct address_space *mapping = inode->i_mapping;
 106         struct page *page;
 107         int rc;
 108
 109         page = grab_cache_page(mapping, index); /* locked page */
 110
 111         if (!IS_ERR(page)) {
 112                 /* Note: Called with "O" and "PAGE_SIZE" this is essentially
 113                  * a no-op for most filesystems, because we write the whole
 114                  * page.  For partial-page I/O this will read in the page.
 115                  */
 116                 rc = mapping->a_ops->prepare_write(NULL, page, 0, PAGE_SIZE);
 117                 if (rc) {
 118                         CERROR("page index %lu, rc = %d\n", index, rc);
 119                         if (rc != -ENOSPC)
 120                                 LBUG();
 121                         GOTO(err_unlock, rc);
 122                 }
 123                 /* XXX not sure if we need this if we are overwriting page */
 124                 if (PageError(page)) {
 125                         CERROR("error on page index %lu, rc = %d\n", index, rc);
 126                         LBUG();
 127                         GOTO(err_unlock, rc = -EIO);
 128                 }
 129         }
 130         return page;
 131
 132 err_unlock:
 133         unlock_page(page);
 134         page_cache_release(page);
 135         return ERR_PTR(rc);
 136 }
 137
 138 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
 139 int wait_on_page_locked(struct page *page)
 140 {
 141         waitfor_one_page(page);
 142         return 0;
 143 }
 144
 145 /* We should only change the file mtime (and not the ctime, like
 146  * update_inode_times() in generic_file_write()) when we only change data. */
 147 static inline void inode_update_time(struct inode *inode, int ctime_too)
 148 {
 149         time_t now = CURRENT_TIME;
 150         if (inode->i_mtime == now && (!ctime_too || inode->i_ctime == now))
 151                 return;
 152         inode->i_mtime = now;
 153         if (ctime_too)
 154                 inode->i_ctime = now;
 155         mark_inode_dirty_sync(inode);
 156 }
 157 #endif
 158
 159 static int lustre_commit_write(struct niobuf_local *lnb)
 160 {
 161         struct page *page = lnb->page;
 162         unsigned from = lnb->offset & ~PAGE_MASK;
 163         unsigned to = from + lnb->len;
 164         struct inode *inode = page->mapping->host;
 165         int err;
 166
 167         LASSERT(to <= PAGE_SIZE);
 168         err = page->mapping->a_ops->commit_write(NULL, page, from, to);
 169 #warning 2.4 folks: wait_on_page_locked does NOT return its error here.
 170         if (!err && IS_SYNC(inode))
 171                 wait_on_page_locked(page);
 172         //SetPageUptodate(page); // the client commit_write will do this
 173
 174         SetPageReferenced(page);
 175         unlock_page(page);
 176         page_cache_release(page);
 177         return err;
 178 }
 179
 180 int filter_get_page_write(struct inode *inode, struct niobuf_local *lnb,
 181                           int *pglocked)
 182 {
 183         unsigned long index = lnb->offset >> PAGE_SHIFT;
 184         struct address_space *mapping = inode->i_mapping;
 185         struct page *page;
 186         int rc;
 187
 188         //ASSERT_PAGE_INDEX(index, GOTO(err, rc = -EINVAL));
 189         if (*pglocked)
 190                 page = grab_cache_page_nowait(mapping, index); /* locked page */
 191         else
 192                 page = grab_cache_page(mapping, index); /* locked page */
 193
 194
 195         /* This page is currently locked, so get a temporary page instead. */
 196         if (page == NULL) {
 197                 CDEBUG(D_INFO, "ino %lu page %ld locked\n", inode->i_ino,index);
 198                 page = alloc_pages(GFP_KERNEL, 0); /* locked page */
 199                 if (page == NULL) {
 200                         CERROR("no memory for a temp page\n");
 201                         GOTO(err, rc = -ENOMEM);
 202                 }
 203                 page->index = index;
 204                 lnb->page = page;
 205                 lnb->flags |= N_LOCAL_TEMP_PAGE;
 206         } else if (!IS_ERR(page)) {
 207                 unsigned from = lnb->offset & ~PAGE_MASK, to = from + lnb->len;
 208                 (*pglocked)++;
 209
 210                 rc = mapping->a_ops->prepare_write(NULL, page, from, to);
 211                 if (rc) {
 212                         if (rc != -ENOSPC)
 213                                 CERROR("page index %lu, rc = %d\n", index, rc);
 214                         GOTO(err_unlock, rc);
 215                 }
 216                 /* XXX not sure if we need this if we are overwriting page */
 217                 if (PageError(page)) {
 218                         CERROR("error on page index %lu, rc = %d\n", index, rc);
 219                         LBUG();
 220                         GOTO(err_unlock, rc = -EIO);
 221                 }
 222                 lnb->page = page;
 223         }
 224
 225         return 0;
 226
 227 err_unlock:
 228         unlock_page(page);
 229         page_cache_release(page);
 230 err:
 231         return lnb->rc = rc;
 232 }
 233
 234 static int filter_preprw_read(int cmd, struct obd_export *exp, struct obdo *oa,
 235                               int objcount, struct obd_ioobj *obj,
 236                               int niocount, struct niobuf_remote *nb,
 237                               struct niobuf_local *res,
 238                               struct obd_trans_info *oti)
 239 {
 240         struct obd_run_ctxt saved;
 241         struct obd_ioobj *o;
 242         struct niobuf_remote *rnb;
 243         struct niobuf_local *lnb;
 244         struct fsfilt_objinfo *fso;
 245         struct dentry *dentry;
 246         struct inode *inode;
 247         int rc = 0, i, j, tot_bytes = 0, cleanup_phase = 0;
 248         unsigned long now = jiffies;
 249         ENTRY;
 250
 251         /* We are currently not supporting multi-obj BRW_READ RPCS at all.
 252          * When we do this function's dentry cleanup will need to be fixed */
 253         LASSERT(objcount == 1);
 254
 255         OBD_ALLOC(fso, objcount * sizeof(*fso));
 256         if (fso == NULL)
 257                 RETURN(-ENOMEM);
 258
 259         memset(res, 0, niocount * sizeof(*res));
 260
 261         push_ctxt(&saved, &exp->exp_obd->u.filter.fo_ctxt, NULL);
 262         for (i = 0, o = obj; i < objcount; i++, o++) {
 263                 struct filter_dentry_data *fdd;
 264                 LASSERT(o->ioo_bufcnt);
 265
 266                 dentry = filter_oa2dentry(exp->exp_obd, oa);
 267                 if (IS_ERR(dentry))
 268                         GOTO(cleanup, rc = PTR_ERR(dentry));
 269
 270                 if (dentry->d_inode == NULL) {
 271                         CERROR("trying to BRW to non-existent file "LPU64"\n",
 272                                o->ioo_id);
 273                         f_dput(dentry);
 274                         GOTO(cleanup, rc = -ENOENT);
 275                 }
 276
 277                 fso[i].fso_dentry = dentry;
 278                 fso[i].fso_bufcnt = o->ioo_bufcnt;
 279
 280                 fdd = dentry->d_fsdata;
 281                 if (fdd == NULL || !atomic_read(&fdd->fdd_open_count))
 282                         CDEBUG(D_PAGE, "I/O to unopened object "LPU64"\n",
 283                                o->ioo_id);
 284         }
 285
 286         if (time_after(jiffies, now + 15 * HZ))
 287                 CERROR("slow prep setup %lus\n", (jiffies - now) / HZ);
 288
 289         for (i = 0, o = obj, rnb = nb, lnb = res; i < objcount; i++, o++) {
 290                 dentry = fso[i].fso_dentry;
 291                 inode = dentry->d_inode;
 292
 293                 for (j = 0; j < o->ioo_bufcnt; j++, rnb++, lnb++) {
 294                         lnb->dentry = dentry;
 295                         lnb->offset = rnb->offset;
 296                         lnb->len    = rnb->len;
 297                         lnb->flags  = rnb->flags;
 298                         lnb->start  = jiffies;
 299
 300                         if (inode->i_size <= rnb->offset) {
 301                                 /* If there's no more data, abort early.
 302                                  * lnb->page == NULL and lnb->rc == 0, so it's
 303                                  * easy to detect later. */
 304                                 lnb->dentry = NULL;
 305                                 break;
 306                         } else {
 307                                 rc = filter_start_page_read(inode, lnb);
 308                         }
 309
 310                         if (rc) {
 311                                 CDEBUG(rc == -ENOSPC ? D_INODE : D_ERROR,
 312                                        "page err %u@"LPU64" %u/%u %p: rc %d\n",
 313                                        lnb->len, lnb->offset, j, o->ioo_bufcnt,
 314                                        dentry, rc);
 315                                 cleanup_phase = 1;
 316                                 GOTO(cleanup, rc);
 317                         }
 318
 319                         tot_bytes += lnb->rc;
 320                         if (lnb->rc < lnb->len) {
 321                                 /* short read, be sure to wait on it */
 322                                 lnb++;
 323                                 break;
 324                         }
 325                 }
 326         }
 327
 328         if (time_after(jiffies, now + 15 * HZ))
 329                 CERROR("slow prep get page %lus\n", (jiffies - now) / HZ);
 330
 331         lprocfs_counter_add(exp->exp_obd->obd_stats, LPROC_FILTER_READ_BYTES,
 332                             tot_bytes);
 333         while (lnb-- > res) {
 334                 rc = filter_finish_page_read(lnb);
 335                 if (rc) {
 336                         CERROR("error page %u@"LPU64" %u %p: rc %d\n", lnb->len,
 337                                lnb->offset, (int)(lnb - res), lnb->dentry, rc);
 338                         cleanup_phase = 1;
 339                         GOTO(cleanup, rc);
 340                 }
 341         }
 342
 343         if (time_after(jiffies, now + 15 * HZ))
 344                 CERROR("slow prep finish page %lus\n", (jiffies - now) / HZ);
 345
 346         EXIT;
 347
 348  cleanup:
 349         switch (cleanup_phase) {
 350         case 1:
 351                 for (lnb = res; lnb < (res + niocount); lnb++) {
 352                         if (lnb->page)
 353                                 page_cache_release(lnb->page);
 354                 }
 355                 f_dput(res->dentry);
 356         case 0:
 357                 OBD_FREE(fso, objcount * sizeof(*fso));
 358                 pop_ctxt(&saved, &exp->exp_obd->u.filter.fo_ctxt, NULL);
 359         }
 360         return rc;
 361 }
 362
 363 /* We need to balance prepare_write() calls with commit_write() calls.
 364  * If the page has been prepared, but we have no data for it, we don't
 365  * want to overwrite valid data on disk, but we still need to zero out
 366  * data for space which was newly allocated.  Like part of what happens
 367  * in __block_prepare_write() for newly allocated blocks.
 368  *
 369  * XXX currently __block_prepare_write() creates buffers for all the
 370  *     pages, and the filesystems mark these buffers as BH_New if they
 371  *     were newly allocated from disk. We use the BH_New flag similarly. */
 372 static int filter_commit_write(struct niobuf_local *lnb, int err)
 373 {
 374 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
 375         if (err) {
 376                 unsigned block_start, block_end;
 377                 struct buffer_head *bh, *head = lnb->page->buffers;
 378                 unsigned blocksize = head->b_size;
 379
 380                 /* debugging: just seeing if this ever happens */
 381                 CDEBUG(err == -ENOSPC ? D_INODE : D_ERROR,
 382                        "called for ino %lu:%lu on err %d\n",
 383                        lnb->page->mapping->host->i_ino, lnb->page->index, err);
 384
 385                 /* Currently one buffer per page, but in the future... */
 386                 for (bh = head, block_start = 0; bh != head || !block_start;
 387                      block_start = block_end, bh = bh->b_this_page) {
 388                         block_end = block_start + blocksize;
 389                         if (buffer_new(bh)) {
 390                                 memset(kmap(lnb->page) + block_start, 0,
 391                                        blocksize);
 392                                 kunmap(lnb->page);
 393                         }
 394                 }
 395         }
 396 #endif
 397         return lustre_commit_write(lnb);
 398 }
 399
 400 /* If we ever start to support multi-object BRW RPCs, we will need to get locks
 401  * on mulitple inodes.  That isn't all, because there still exists the
 402  * possibility of a truncate starting a new transaction while holding the ext3
 403  * rwsem = write while some writes (which have started their transactions here)
 404  * blocking on the ext3 rwsem = read => lock inversion.
 405  *
 406  * The handling gets very ugly when dealing with locked pages.  It may be easier
 407  * to just get rid of the locked page code (which has problems of its own) and
 408  * either discover we do not need it anymore (i.e. it was a symptom of another
 409  * bug) or ensure we get the page locks in an appropriate order. */
 410 static int filter_preprw_write(int cmd, struct obd_export *exp, struct obdo *oa,
 411                                int objcount, struct obd_ioobj *obj,
 412                                int niocount, struct niobuf_remote *nb,
 413                                struct niobuf_local *res,
 414                                struct obd_trans_info *oti)
 415 {
 416         struct obd_run_ctxt saved;
 417         struct obd_ioobj *o;
 418         struct niobuf_remote *rnb;
 419         struct niobuf_local *lnb;
 420         struct fsfilt_objinfo *fso;
 421         struct dentry *dentry;
 422         int pglocked = 0, rc = 0, i, j, tot_bytes = 0;
 423         unsigned long now = jiffies;
 424         ENTRY;
 425         LASSERT(objcount == 1);
 426
 427         OBD_ALLOC(fso, objcount * sizeof(*fso));
 428         if (fso == NULL)
 429                 RETURN(-ENOMEM);
 430
 431         memset(res, 0, niocount * sizeof(*res));
 432
 433         push_ctxt(&saved, &exp->exp_obd->u.filter.fo_ctxt, NULL);
 434         for (i = 0, o = obj; i < objcount; i++, o++) {
 435                 struct filter_dentry_data *fdd;
 436                 LASSERT(o->ioo_bufcnt);
 437
 438                 dentry = filter_oa2dentry(exp->exp_obd, oa);
 439                 if (IS_ERR(dentry))
 440                         GOTO(out_objinfo, rc = PTR_ERR(dentry));
 441
 442                 if (dentry->d_inode == NULL) {
 443                         CERROR("trying to BRW to non-existent file "LPU64"\n",
 444                                o->ioo_id);
 445                         f_dput(dentry);
 446                         GOTO(out_objinfo, rc = -ENOENT);
 447                 }
 448
 449                 fso[i].fso_dentry = dentry;
 450                 fso[i].fso_bufcnt = o->ioo_bufcnt;
 451
 452                 down(&dentry->d_inode->i_sem);
 453                 fdd = dentry->d_fsdata;
 454                 if (fdd == NULL || !atomic_read(&fdd->fdd_open_count))
 455                         CDEBUG(D_PAGE, "I/O to unopened object "LPU64"\n",
 456                                o->ioo_id);
 457         }
 458
 459         if (time_after(jiffies, now + 15 * HZ))
 460                 CERROR("slow prep setup %lus\n", (jiffies - now) / HZ);
 461
 462         LASSERT(oti != NULL);
 463         oti->oti_handle = fsfilt_brw_start(exp->exp_obd, objcount, fso,
 464                                            niocount, oti);
 465         if (IS_ERR(oti->oti_handle)) {
 466                 rc = PTR_ERR(oti->oti_handle);
 467                 CDEBUG(rc == -ENOSPC ? D_INODE : D_ERROR,
 468                        "error starting transaction: rc = %d\n", rc);
 469                 oti->oti_handle = NULL;
 470                 GOTO(out_objinfo, rc);
 471         }
 472
 473         for (i = 0, o = obj, rnb = nb, lnb = res; i < objcount; i++, o++) {
 474                 dentry = fso[i].fso_dentry;
 475                 for (j = 0; j < o->ioo_bufcnt; j++, rnb++, lnb++) {
 476                         if (j == 0)
 477                                 lnb->dentry = dentry;
 478                         else
 479                                 lnb->dentry = dget(dentry);
 480
 481                         lnb->offset = rnb->offset;
 482                         lnb->len    = rnb->len;
 483                         lnb->flags  = rnb->flags;
 484                         lnb->start  = jiffies;
 485
 486                         rc = filter_get_page_write(dentry->d_inode, lnb,
 487                                                    &pglocked);
 488                         if (rc)
 489                                 up(&dentry->d_inode->i_sem);
 490
 491                         if (rc) {
 492                                 CDEBUG(rc == -ENOSPC ? D_INODE : D_ERROR,
 493                                        "page err %u@"LPU64" %u/%u %p: rc %d\n",
 494                                        lnb->len, lnb->offset, j, o->ioo_bufcnt,
 495                                        dentry, rc);
 496                                 f_dput(dentry);
 497                                 GOTO(out_pages, rc);
 498                         }
 499                         tot_bytes += lnb->len;
 500                 }
 501         }
 502
 503         if (time_after(jiffies, now + 15 * HZ))
 504                 CERROR("slow prep get page %lus\n", (jiffies - now) / HZ);
 505
 506         lprocfs_counter_add(exp->exp_obd->obd_stats, LPROC_FILTER_WRITE_BYTES,
 507                             tot_bytes);
 508
 509         EXIT;
 510 out:
 511         OBD_FREE(fso, objcount * sizeof(*fso));
 512         /* we saved the journal handle into oti->oti_handle instead */
 513         current->journal_info = NULL;
 514         pop_ctxt(&saved, &exp->exp_obd->u.filter.fo_ctxt, NULL);
 515         return rc;
 516
 517 out_pages:
 518         while (lnb-- > res) {
 519                 filter_commit_write(lnb, rc);
 520                 up(&lnb->dentry->d_inode->i_sem);
 521                 f_dput(lnb->dentry);
 522         }
 523         filter_finish_transno(exp, oti, rc);
 524         fsfilt_commit(exp->exp_obd,
 525                       filter_parent(exp->exp_obd,S_IFREG,obj->ioo_id)->d_inode,
 526                       oti->oti_handle, 0);
 527         goto out; /* dropped the dentry refs already (one per page) */
 528
 529 out_objinfo:
 530         for (i = 0; i < objcount && fso[i].fso_dentry; i++) {
 531                 up(&fso[i].fso_dentry->d_inode->i_sem);
 532                 f_dput(fso[i].fso_dentry);
 533         }
 534         goto out;
 535 }
 536
 537 int filter_preprw(int cmd, struct obd_export *exp, struct obdo *oa,
 538                   int objcount, struct obd_ioobj *obj, int niocount,
 539                   struct niobuf_remote *nb, struct niobuf_local *res,
 540                   struct obd_trans_info *oti)
 541 {
 542         if (cmd == OBD_BRW_WRITE)
 543                 return filter_preprw_write(cmd, exp, oa, objcount, obj,
 544                                            niocount, nb, res, oti);
 545
 546         if (cmd == OBD_BRW_READ)
 547                 return filter_preprw_read(cmd, exp, oa, objcount, obj,
 548                                           niocount, nb, res, oti);
 549
 550         LBUG();
 551
 552         return -EPROTO;
 553 }
 554
 555 /* It is highly unlikely that we would ever get an error here.  The page we want
 556  * to get was previously locked, so it had to have already allocated the space,
 557  * and we were just writing over the same data, so there would be no hole in the
 558  * file.
 559  *
 560  * XXX: possibility of a race with truncate could exist, need to check that.
 561  *      There are no guarantees w.r.t. write order even on a local filesystem,
 562  *      although the normal response would be to return the number of bytes
 563  *      successfully written and leave the rest to the app. */
 564 static int filter_write_locked_page(struct niobuf_local *lnb)
 565 {
 566         struct page *lpage;
 567         void *lpage_addr, *lnb_addr;
 568         int rc;
 569         ENTRY;
 570
 571         lpage = lustre_get_page_write(lnb->dentry->d_inode, lnb->page->index);
 572         if (IS_ERR(lpage)) {
 573                 rc = PTR_ERR(lpage);
 574                 CERROR("error getting locked page index %ld: rc = %d\n",
 575                        lnb->page->index, rc);
 576                 LBUG();
 577                 lustre_commit_write(lnb);
 578                 RETURN(rc);
 579         }
 580
 581         /* 2 kmaps == vanishingly small deadlock opportunity */
 582         lpage_addr = kmap(lpage);
 583         lnb_addr = kmap(lnb->page);
 584
 585         memcpy(lpage_addr, lnb_addr, PAGE_SIZE);
 586
 587         kunmap(lnb->page);
 588         kunmap(lpage);
 589
 590         page_cache_release(lnb->page);
 591
 592         lnb->page = lpage;
 593         rc = lustre_commit_write(lnb);
 594         if (rc)
 595                 CERROR("error committing locked page %ld: rc = %d\n",
 596                        lnb->page->index, rc);
 597         RETURN(rc);
 598 }
 599
 600 static int filter_commitrw_read(struct obd_export *exp, int objcount,
 601                                 struct obd_ioobj *obj, int niocount,
 602                                 struct niobuf_local *res,
 603                                 struct obd_trans_info *oti)
 604 {
 605         struct obd_ioobj *o;
 606         struct niobuf_local *lnb;
 607         int i, j;
 608         ENTRY;
 609
 610         for (i = 0, o = obj, lnb = res; i < objcount; i++, o++) {
 611                 for (j = 0 ; j < o->ioo_bufcnt ; j++, lnb++) {
 612                         if (lnb->page != NULL)
 613                                 page_cache_release(lnb->page);
 614                 }
 615         }
 616         f_dput(res->dentry);
 617         RETURN(0);
 618 }
 619
 620 static int
 621 filter_commitrw_write(int cmd, struct obd_export *exp, struct obdo *oa,
 622                       int objcount, struct obd_ioobj *obj, int niocount,
 623                       struct niobuf_local *res, struct obd_trans_info *oti)
 624 {
 625         struct obd_run_ctxt saved;
 626         struct obd_ioobj *o;
 627         struct niobuf_local *lnb;
 628         struct obd_device *obd = exp->exp_obd;
 629         int found_locked = 0, rc = 0, i;
 630         int nested_trans = current->journal_info != NULL;
 631         unsigned long now = jiffies;  /* DEBUGGING OST TIMEOUTS */
 632         ENTRY;
 633
 634         push_ctxt(&saved, &obd->u.filter.fo_ctxt, NULL);
 635
 636         if (cmd & OBD_BRW_WRITE) {
 637                 LASSERT(oti);
 638                 LASSERT(current->journal_info == NULL ||
 639                         current->journal_info == oti->oti_handle);
 640                 current->journal_info = oti->oti_handle;
 641         }
 642
 643         for (i = 0, o = obj, lnb = res; i < objcount; i++, o++) {
 644                 struct inode *inode;
 645                 int j;
 646
 647                 /* If all of the page reads were beyond EOF, let's pretend
 648                  * this read didn't really happen at all. */
 649                 if (lnb->dentry == NULL) {
 650                         oa->o_valid = OBD_MD_FLID|(oa->o_valid&OBD_MD_FLCKSUM);
 651                         continue;
 652                 }
 653
 654                 inode = igrab(lnb->dentry->d_inode);
 655
 656                 if (cmd & OBD_BRW_WRITE) {
 657                         /* FIXME: MULTI OBJECT BRW */
 658                         if (oa && oa->o_valid & (OBD_MD_FLMTIME|OBD_MD_FLCTIME))
 659                                 obdo_refresh_inode(inode, oa, OBD_MD_FLATIME |
 660                                                    OBD_MD_FLMTIME |
 661                                                    OBD_MD_FLCTIME);
 662                         else
 663                                 inode_update_time(lnb->dentry->d_inode, 1);
 664                 } else if (oa && oa->o_valid & OBD_MD_FLATIME) {
 665                         /* Note that we don't necessarily write this to disk */
 666                         obdo_refresh_inode(inode, oa, OBD_MD_FLATIME);
 667                 }
 668
 669                 for (j = 0 ; j < o->ioo_bufcnt ; j++, lnb++) {
 670                         if (lnb->page == NULL) {
 671                                 continue;
 672                         }
 673
 674                         if (lnb->flags & N_LOCAL_TEMP_PAGE) {
 675                                 found_locked++;
 676                                 continue;
 677                         }
 678
 679                         if (time_after(jiffies, lnb->start + 15 * HZ))
 680                                 CERROR("slow commitrw %lusi (%lus)\n",
 681                                        (jiffies - lnb->start) / HZ,
 682                                        (jiffies - now) / HZ);
 683
 684                         if (cmd & OBD_BRW_WRITE) {
 685                                 int err = filter_commit_write(lnb, 0);
 686
 687                                 if (!rc)
 688                                         rc = err;
 689                         } else {
 690                                 page_cache_release(lnb->page);
 691                         }
 692
 693                         f_dput(lnb->dentry);
 694                         if (time_after(jiffies, lnb->start + 15 * HZ))
 695                                 CERROR("slow commit_write %lus (%lus)\n",
 696                                        (jiffies - lnb->start) / HZ,
 697                                        (jiffies - now) / HZ);
 698                 }
 699
 700                 /* FIXME: MULTI OBJECT BRW */
 701                 if (oa) {
 702                         oa->o_valid = OBD_MD_FLID|(oa->o_valid&OBD_MD_FLCKSUM);
 703                         obdo_from_inode(oa, inode, FILTER_VALID_FLAGS);
 704                 }
 705
 706                 if (cmd & OBD_BRW_WRITE)
 707                         up(&inode->i_sem);
 708
 709                 iput(inode);
 710         }
 711
 712         for (i = 0, o = obj, lnb = res; found_locked > 0 && i < objcount;
 713              i++, o++) {
 714                 int j;
 715
 716                 for (j = 0 ; j < o->ioo_bufcnt ; j++, lnb++) {
 717                         int err;
 718                         if (!(lnb->flags & N_LOCAL_TEMP_PAGE))
 719                                 continue;
 720
 721                         if (time_after(jiffies, lnb->start + 15 * HZ))
 722                                 CERROR("slow commitrw locked %lus (%lus)\n",
 723                                        (jiffies - lnb->start) / HZ,
 724                                        (jiffies - now) / HZ);
 725
 726                         err = filter_write_locked_page(lnb);
 727                         if (!rc)
 728                                 rc = err;
 729                         f_dput(lnb->dentry);
 730                         found_locked--;
 731
 732                         if (time_after(jiffies, lnb->start + 15 * HZ))
 733                                 CERROR("slow commit_write locked %lus (%lus)\n",
 734                                        (jiffies - lnb->start) / HZ,
 735                                        (jiffies - now) / HZ);
 736                 }
 737         }
 738
 739         if (cmd & OBD_BRW_WRITE) {
 740                 /* We just want any dentry for the commit, for now */
 741                 struct dentry *dparent = filter_parent(obd, S_IFREG, 0);
 742                 int err;
 743
 744                 rc = filter_finish_transno(exp, oti, rc);
 745                 err = fsfilt_commit(obd, dparent->d_inode, oti->oti_handle,
 746                                     obd_sync_filter);
 747                 if (err)
 748                         rc = err;
 749                 if (obd_sync_filter)
 750                         LASSERT(oti->oti_transno <= obd->obd_last_committed);
 751                 if (time_after(jiffies, now + 15 * HZ))
 752                         CERROR("slow commitrw commit %lus\n", (jiffies-now)/HZ);
 753         }
 754
 755         LASSERT(nested_trans || current->journal_info == NULL);
 756         pop_ctxt(&saved, &obd->u.filter.fo_ctxt, NULL);
 757         RETURN(rc);
 758 }
 759
 760 /* XXX needs to trickle its oa down */
 761 int filter_commitrw(int cmd, struct obd_export *exp, struct obdo *oa,
 762                     int objcount, struct obd_ioobj *obj, int niocount,
 763                     struct niobuf_local *res, struct obd_trans_info *oti)
 764 {
 765         if (cmd == OBD_BRW_WRITE)
 766                 return filter_commitrw_write(cmd, exp, oa, objcount, obj,
 767                                              niocount, res, oti);
 768         if (cmd == OBD_BRW_READ)
 769                 return filter_commitrw_read(exp, objcount, obj, niocount,
 770                                             res, oti);
 771         LBUG();
 772         return -EPROTO;
 773 }
 774
 775 int filter_brw(int cmd, struct lustre_handle *conn, struct obdo *oa,
 776                struct lov_stripe_md *lsm, obd_count oa_bufs,
 777                struct brw_page *pga, struct obd_trans_info *oti)
 778 {
 779         struct obd_export *exp;
 780         struct obd_ioobj ioo;
 781         struct niobuf_local *lnb;
 782         struct niobuf_remote *rnb;
 783         obd_count i;
 784         int ret = 0;
 785         ENTRY;
 786
 787         exp = class_conn2export(conn);
 788         if (exp == NULL) {
 789                 CDEBUG(D_IOCTL, "invalid client cookie "LPX64"\n",conn->cookie);
 790                 RETURN(-EINVAL);
 791         }
 792
 793         OBD_ALLOC(lnb, oa_bufs * sizeof(struct niobuf_local));
 794         OBD_ALLOC(rnb, oa_bufs * sizeof(struct niobuf_remote));
 795
 796         if (lnb == NULL || rnb == NULL)
 797                 GOTO(out, ret = -ENOMEM);
 798
 799         for (i = 0; i < oa_bufs; i++) {
 800                 rnb[i].offset = pga[i].off;
 801                 rnb[i].len = pga[i].count;
 802         }
 803
 804         ioo.ioo_id = oa->o_id;
 805         ioo.ioo_gr = 0;
 806         ioo.ioo_type = oa->o_mode & S_IFMT;
 807         ioo.ioo_bufcnt = oa_bufs;
 808
 809         ret = filter_preprw(cmd, exp, oa, 1, &ioo, oa_bufs, rnb, lnb, oti);
 810         if (ret != 0)
 811                 GOTO(out, ret);
 812
 813         for (i = 0; i < oa_bufs; i++) {
 814                 void *virt = kmap(pga[i].pg);
 815                 obd_off off = pga[i].off & ~PAGE_MASK;
 816                 void *addr = kmap(lnb[i].page);
 817
 818                 /* 2 kmaps == vanishingly small deadlock opportunity */
 819
 820                 if (cmd & OBD_BRW_WRITE)
 821                         memcpy(addr + off, virt + off, pga[i].count);
 822                 else
 823                         memcpy(virt + off, addr + off, pga[i].count);
 824
 825                 kunmap(addr);
 826                 kunmap(virt);
 827         }
 828
 829         ret = filter_commitrw(cmd, exp, oa, 1, &ioo, oa_bufs, lnb, oti);
 830
 831 out:
 832         if (lnb)
 833                 OBD_FREE(lnb, oa_bufs * sizeof(struct niobuf_local));
 834         if (rnb)
 835                 OBD_FREE(rnb, oa_bufs * sizeof(struct niobuf_remote));
 836         class_export_put(exp);
 837         RETURN(ret);
 838 }