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