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                                 break;
 305                         } else {
 306                                 rc = filter_start_page_read(inode, lnb);
 307                         }
 308
 309                         if (rc) {
 310                                 CDEBUG(rc == -ENOSPC ? D_INODE : D_ERROR,
 311                                        "page err %u@"LPU64" %u/%u %p: rc %d\n",
 312                                        lnb->len, lnb->offset, j, o->ioo_bufcnt,
 313                                        dentry, rc);
 314                                 cleanup_phase = 1;
 315                                 GOTO(cleanup, rc);
 316                         }
 317
 318                         tot_bytes += lnb->rc;
 319                         if (lnb->rc < lnb->len) {
 320                                 /* short read, be sure to wait on it */
 321                                 lnb++;
 322                                 break;
 323                         }
 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                         cleanup_phase = 1;
 338                         GOTO(cleanup, 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
 347  cleanup:
 348         switch (cleanup_phase) {
 349         case 1:
 350                 for (lnb = res; lnb < (res + niocount); lnb++) {
 351                         if (lnb->page)
 352                                 page_cache_release(lnb->page);
 353                 }
 354                 if (res->dentry != NULL)
 355                         f_dput(res->dentry);
 356                 else
 357                         CERROR("NULL dentry in cleanup -- tell CFS\n");
 358                 res->dentry = NULL;
 359         case 0:
 360                 OBD_FREE(fso, objcount * sizeof(*fso));
 361                 pop_ctxt(&saved, &exp->exp_obd->u.filter.fo_ctxt, NULL);
 362         }
 363         return rc;
 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         /* We should never be called during a transaction */
 431         LASSERT(oti != NULL);
 432         LASSERT(oti->oti_handle == NULL);
 433         LASSERT(current->journal_info == NULL);
 434
 435         OBD_ALLOC(fso, objcount * sizeof(*fso));
 436         if (fso == NULL)
 437                 RETURN(-ENOMEM);
 438
 439         memset(res, 0, niocount * sizeof(*res));
 440
 441         push_ctxt(&saved, &exp->exp_obd->u.filter.fo_ctxt, NULL);
 442         for (i = 0, o = obj; i < objcount; i++, o++) {
 443                 struct filter_dentry_data *fdd;
 444                 LASSERT(o->ioo_bufcnt);
 445
 446                 dentry = filter_oa2dentry(exp->exp_obd, oa);
 447                 if (IS_ERR(dentry))
 448                         GOTO(out_objinfo, rc = PTR_ERR(dentry));
 449
 450                 if (dentry->d_inode == NULL) {
 451                         CERROR("trying to BRW to non-existent file "LPU64"\n",
 452                                o->ioo_id);
 453                         f_dput(dentry);
 454                         GOTO(out_objinfo, rc = -ENOENT);
 455                 }
 456
 457                 fso[i].fso_dentry = dentry;
 458                 fso[i].fso_bufcnt = o->ioo_bufcnt;
 459
 460                 down(&dentry->d_inode->i_sem);
 461                 fdd = dentry->d_fsdata;
 462                 if (fdd == NULL || !atomic_read(&fdd->fdd_open_count))
 463                         CDEBUG(D_PAGE, "I/O to unopened object "LPU64"\n",
 464                                o->ioo_id);
 465         }
 466
 467         if (time_after(jiffies, now + 15 * HZ))
 468                 CERROR("slow prep setup %lus\n", (jiffies - now) / HZ);
 469
 470         oti->oti_handle = fsfilt_brw_start(exp->exp_obd, objcount, fso,
 471                                            niocount, oti);
 472         if (IS_ERR(oti->oti_handle)) {
 473                 rc = PTR_ERR(oti->oti_handle);
 474                 CDEBUG(rc == -ENOSPC ? D_INODE : D_ERROR,
 475                        "error starting transaction: rc = %d\n", rc);
 476                 oti->oti_handle = NULL;
 477                 GOTO(out_objinfo, rc);
 478         }
 479
 480         for (i = 0, o = obj, rnb = nb, lnb = res; i < objcount; i++, o++) {
 481                 dentry = fso[i].fso_dentry;
 482                 for (j = 0; j < o->ioo_bufcnt; j++, rnb++, lnb++) {
 483                         if (j == 0)
 484                                 lnb->dentry = dentry;
 485                         else
 486                                 lnb->dentry = dget(dentry);
 487
 488                         lnb->offset = rnb->offset;
 489                         lnb->len    = rnb->len;
 490                         lnb->flags  = rnb->flags;
 491                         lnb->start  = jiffies;
 492
 493                         rc = filter_get_page_write(dentry->d_inode, lnb,
 494                                                    &pglocked);
 495                         if (rc)
 496                                 up(&dentry->d_inode->i_sem);
 497
 498                         if (rc) {
 499                                 CDEBUG(rc == -ENOSPC ? D_INODE : D_ERROR,
 500                                        "page err %u@"LPU64" %u/%u %p: rc %d\n",
 501                                        lnb->len, lnb->offset, j, o->ioo_bufcnt,
 502                                        dentry, rc);
 503                                 f_dput(dentry);
 504                                 GOTO(out_pages, rc);
 505                         }
 506                         tot_bytes += lnb->len;
 507                 }
 508         }
 509
 510         if (time_after(jiffies, now + 15 * HZ))
 511                 CERROR("slow prep get page %lus\n", (jiffies - now) / HZ);
 512
 513         lprocfs_counter_add(exp->exp_obd->obd_stats, LPROC_FILTER_WRITE_BYTES,
 514                             tot_bytes);
 515
 516         EXIT;
 517 out:
 518         OBD_FREE(fso, objcount * sizeof(*fso));
 519         /* we saved the journal handle into oti->oti_handle instead */
 520         current->journal_info = NULL;
 521         pop_ctxt(&saved, &exp->exp_obd->u.filter.fo_ctxt, NULL);
 522         return rc;
 523
 524 out_pages:
 525         while (lnb-- > res) {
 526                 filter_commit_write(lnb, rc);
 527                 up(&lnb->dentry->d_inode->i_sem);
 528                 f_dput(lnb->dentry);
 529         }
 530         filter_finish_transno(exp, oti, rc);
 531         fsfilt_commit(exp->exp_obd,
 532                       filter_parent(exp->exp_obd,S_IFREG,obj->ioo_id)->d_inode,
 533                       oti->oti_handle, 0);
 534         LASSERT(current->journal_info == NULL);
 535         oti->oti_handle = NULL;
 536         goto out; /* dropped the dentry refs already (one per page) */
 537
 538 out_objinfo:
 539         for (i = 0; i < objcount && fso[i].fso_dentry; i++) {
 540                 up(&fso[i].fso_dentry->d_inode->i_sem);
 541                 f_dput(fso[i].fso_dentry);
 542         }
 543         goto out;
 544 }
 545
 546 int filter_preprw(int cmd, struct obd_export *exp, struct obdo *oa,
 547                   int objcount, struct obd_ioobj *obj, int niocount,
 548                   struct niobuf_remote *nb, struct niobuf_local *res,
 549                   struct obd_trans_info *oti)
 550 {
 551         if (cmd == OBD_BRW_WRITE)
 552                 return filter_preprw_write(cmd, exp, oa, objcount, obj,
 553                                            niocount, nb, res, oti);
 554
 555         if (cmd == OBD_BRW_READ)
 556                 return filter_preprw_read(cmd, exp, oa, objcount, obj,
 557                                           niocount, nb, res, oti);
 558
 559         LBUG();
 560
 561         return -EPROTO;
 562 }
 563
 564 /* It is highly unlikely that we would ever get an error here.  The page we want
 565  * to get was previously locked, so it had to have already allocated the space,
 566  * and we were just writing over the same data, so there would be no hole in the
 567  * file.
 568  *
 569  * XXX: possibility of a race with truncate could exist, need to check that.
 570  *      There are no guarantees w.r.t. write order even on a local filesystem,
 571  *      although the normal response would be to return the number of bytes
 572  *      successfully written and leave the rest to the app. */
 573 static int filter_write_locked_page(struct niobuf_local *lnb)
 574 {
 575         struct page *lpage;
 576         void *lpage_addr, *lnb_addr;
 577         int rc;
 578         ENTRY;
 579
 580         lpage = lustre_get_page_write(lnb->dentry->d_inode, lnb->page->index);
 581         if (IS_ERR(lpage)) {
 582                 rc = PTR_ERR(lpage);
 583                 CERROR("error getting locked page index %ld: rc = %d\n",
 584                        lnb->page->index, rc);
 585                 LBUG();
 586                 lustre_commit_write(lnb);
 587                 RETURN(rc);
 588         }
 589
 590         /* 2 kmaps == vanishingly small deadlock opportunity */
 591         lpage_addr = kmap(lpage);
 592         lnb_addr = kmap(lnb->page);
 593
 594         memcpy(lpage_addr, lnb_addr, PAGE_SIZE);
 595
 596         kunmap(lnb->page);
 597         kunmap(lpage);
 598
 599         page_cache_release(lnb->page);
 600
 601         lnb->page = lpage;
 602         rc = lustre_commit_write(lnb);
 603         if (rc)
 604                 CERROR("error committing locked page %ld: rc = %d\n",
 605                        lnb->page->index, rc);
 606         RETURN(rc);
 607 }
 608
 609 static int filter_commitrw_read(struct obd_export *exp, int objcount,
 610                                 struct obd_ioobj *obj, int niocount,
 611                                 struct niobuf_local *res,
 612                                 struct obd_trans_info *oti)
 613 {
 614         struct obd_ioobj *o;
 615         struct niobuf_local *lnb;
 616         int i, j;
 617         ENTRY;
 618
 619         for (i = 0, o = obj, lnb = res; i < objcount; i++, o++) {
 620                 for (j = 0 ; j < o->ioo_bufcnt ; j++, lnb++) {
 621                         if (lnb->page != NULL)
 622                                 page_cache_release(lnb->page);
 623                 }
 624         }
 625         if (res->dentry != NULL)
 626                 f_dput(res->dentry);
 627         RETURN(0);
 628 }
 629
 630 static int
 631 filter_commitrw_write(int cmd, struct obd_export *exp, struct obdo *oa,
 632                       int objcount, struct obd_ioobj *obj, int niocount,
 633                       struct niobuf_local *res, struct obd_trans_info *oti)
 634 {
 635         struct obd_run_ctxt saved;
 636         struct obd_ioobj *o;
 637         struct niobuf_local *lnb;
 638         struct obd_device *obd = exp->exp_obd;
 639         int found_locked = 0, rc = 0, i;
 640         unsigned long now = jiffies;  /* DEBUGGING OST TIMEOUTS */
 641         ENTRY;
 642
 643         push_ctxt(&saved, &obd->u.filter.fo_ctxt, NULL);
 644
 645         LASSERT(current->journal_info == NULL);
 646         if (cmd & OBD_BRW_WRITE) {
 647                 LASSERT(oti != NULL);
 648                 LASSERT(oti->oti_handle != NULL);
 649                 current->journal_info = oti->oti_handle;
 650         }
 651
 652         for (i = 0, o = obj, lnb = res; i < objcount; i++, o++) {
 653                 struct inode *inode;
 654                 int j;
 655
 656                 /* If all of the page reads were beyond EOF, let's pretend
 657                  * this read didn't really happen at all. */
 658                 if (lnb->dentry == NULL) {
 659                         oa->o_valid = OBD_MD_FLID|(oa->o_valid&OBD_MD_FLCKSUM);
 660                         continue;
 661                 }
 662
 663                 inode = igrab(lnb->dentry->d_inode);
 664
 665                 if (cmd & OBD_BRW_WRITE) {
 666                         /* FIXME: MULTI OBJECT BRW */
 667                         if (oa && oa->o_valid & (OBD_MD_FLMTIME|OBD_MD_FLCTIME))
 668                                 obdo_refresh_inode(inode, oa, OBD_MD_FLATIME |
 669                                                    OBD_MD_FLMTIME |
 670                                                    OBD_MD_FLCTIME);
 671                         else
 672                                 inode_update_time(lnb->dentry->d_inode, 1);
 673                 } else if (oa && oa->o_valid & OBD_MD_FLATIME) {
 674                         /* Note that we don't necessarily write this to disk */
 675                         obdo_refresh_inode(inode, oa, OBD_MD_FLATIME);
 676                 }
 677
 678                 for (j = 0 ; j < o->ioo_bufcnt ; j++, lnb++) {
 679                         if (lnb->page == NULL) {
 680                                 continue;
 681                         }
 682
 683                         if (lnb->flags & N_LOCAL_TEMP_PAGE) {
 684                                 found_locked++;
 685                                 continue;
 686                         }
 687
 688                         if (time_after(jiffies, lnb->start + 15 * HZ))
 689                                 CERROR("slow commitrw %lusi (%lus)\n",
 690                                        (jiffies - lnb->start) / HZ,
 691                                        (jiffies - now) / HZ);
 692
 693                         if (cmd & OBD_BRW_WRITE) {
 694                                 int err = filter_commit_write(lnb, 0);
 695
 696                                 if (!rc)
 697                                         rc = err;
 698                         } else {
 699                                 page_cache_release(lnb->page);
 700                         }
 701
 702                         f_dput(lnb->dentry);
 703                         if (time_after(jiffies, lnb->start + 15 * HZ))
 704                                 CERROR("slow commit_write %lus (%lus)\n",
 705                                        (jiffies - lnb->start) / HZ,
 706                                        (jiffies - now) / HZ);
 707                 }
 708
 709                 /* FIXME: MULTI OBJECT BRW */
 710                 if (oa) {
 711                         oa->o_valid = OBD_MD_FLID|(oa->o_valid&OBD_MD_FLCKSUM);
 712                         obdo_from_inode(oa, inode, FILTER_VALID_FLAGS);
 713                 }
 714
 715                 if (cmd & OBD_BRW_WRITE)
 716                         up(&inode->i_sem);
 717
 718                 iput(inode);
 719         }
 720
 721         for (i = 0, o = obj, lnb = res; found_locked > 0 && i < objcount;
 722              i++, o++) {
 723                 int j;
 724
 725                 for (j = 0 ; j < o->ioo_bufcnt ; j++, lnb++) {
 726                         int err;
 727                         if (!(lnb->flags & N_LOCAL_TEMP_PAGE))
 728                                 continue;
 729
 730                         if (time_after(jiffies, lnb->start + 15 * HZ))
 731                                 CERROR("slow commitrw locked %lus (%lus)\n",
 732                                        (jiffies - lnb->start) / HZ,
 733                                        (jiffies - now) / HZ);
 734
 735                         err = filter_write_locked_page(lnb);
 736                         if (!rc)
 737                                 rc = err;
 738                         f_dput(lnb->dentry);
 739                         found_locked--;
 740
 741                         if (time_after(jiffies, lnb->start + 15 * HZ))
 742                                 CERROR("slow commit_write locked %lus (%lus)\n",
 743                                        (jiffies - lnb->start) / HZ,
 744                                        (jiffies - now) / HZ);
 745                 }
 746         }
 747
 748         if (cmd & OBD_BRW_WRITE) {
 749                 /* We just want any dentry for the commit, for now */
 750                 struct dentry *dparent = filter_parent(obd, S_IFREG, 0);
 751                 int err;
 752
 753                 rc = filter_finish_transno(exp, oti, rc);
 754                 err = fsfilt_commit(obd, dparent->d_inode, oti->oti_handle,
 755                                     obd_sync_filter);
 756                 LASSERT(current->journal_info == NULL);
 757                 oti->oti_handle = NULL;
 758                 if (err)
 759                         rc = err;
 760                 if (obd_sync_filter)
 761                         LASSERT(oti->oti_transno <= obd->obd_last_committed);
 762                 if (time_after(jiffies, now + 15 * HZ))
 763                         CERROR("slow commitrw commit %lus\n", (jiffies-now)/HZ);
 764         }
 765
 766         pop_ctxt(&saved, &obd->u.filter.fo_ctxt, NULL);
 767         RETURN(rc);
 768 }
 769
 770 /* XXX needs to trickle its oa down */
 771 int filter_commitrw(int cmd, struct obd_export *exp, struct obdo *oa,
 772                     int objcount, struct obd_ioobj *obj, int niocount,
 773                     struct niobuf_local *res, struct obd_trans_info *oti)
 774 {
 775         if (cmd == OBD_BRW_WRITE)
 776                 return filter_commitrw_write(cmd, exp, oa, objcount, obj,
 777                                              niocount, res, oti);
 778         if (cmd == OBD_BRW_READ)
 779                 return filter_commitrw_read(exp, objcount, obj, niocount,
 780                                             res, oti);
 781         LBUG();
 782         return -EPROTO;
 783 }
 784
 785 int filter_brw(int cmd, struct lustre_handle *conn, struct obdo *oa,
 786                struct lov_stripe_md *lsm, obd_count oa_bufs,
 787                struct brw_page *pga, struct obd_trans_info *oti)
 788 {
 789         struct obd_export *exp;
 790         struct obd_ioobj ioo;
 791         struct niobuf_local *lnb;
 792         struct niobuf_remote *rnb;
 793         obd_count i;
 794         int ret = 0;
 795         ENTRY;
 796
 797         exp = class_conn2export(conn);
 798         if (exp == NULL) {
 799                 CDEBUG(D_IOCTL, "invalid client cookie "LPX64"\n",conn->cookie);
 800                 RETURN(-EINVAL);
 801         }
 802
 803         OBD_ALLOC(lnb, oa_bufs * sizeof(struct niobuf_local));
 804         OBD_ALLOC(rnb, oa_bufs * sizeof(struct niobuf_remote));
 805
 806         if (lnb == NULL || rnb == NULL)
 807                 GOTO(out, ret = -ENOMEM);
 808
 809         for (i = 0; i < oa_bufs; i++) {
 810                 rnb[i].offset = pga[i].off;
 811                 rnb[i].len = pga[i].count;
 812         }
 813
 814         ioo.ioo_id = oa->o_id;
 815         ioo.ioo_gr = 0;
 816         ioo.ioo_type = oa->o_mode & S_IFMT;
 817         ioo.ioo_bufcnt = oa_bufs;
 818
 819         ret = filter_preprw(cmd, exp, oa, 1, &ioo, oa_bufs, rnb, lnb, oti);
 820         if (ret != 0)
 821                 GOTO(out, ret);
 822
 823         for (i = 0; i < oa_bufs; i++) {
 824                 void *virt = kmap(pga[i].pg);
 825                 obd_off off = pga[i].off & ~PAGE_MASK;
 826                 void *addr = kmap(lnb[i].page);
 827
 828                 /* 2 kmaps == vanishingly small deadlock opportunity */
 829
 830                 if (cmd & OBD_BRW_WRITE)
 831                         memcpy(addr + off, virt + off, pga[i].count);
 832                 else
 833                         memcpy(virt + off, addr + off, pga[i].count);
 834
 835                 kunmap(addr);
 836                 kunmap(virt);
 837         }
 838
 839         ret = filter_commitrw(cmd, exp, oa, 1, &ioo, oa_bufs, lnb, oti);
 840
 841 out:
 842         if (lnb)
 843                 OBD_FREE(lnb, oa_bufs * sizeof(struct niobuf_local));
 844         if (rnb)
 845                 OBD_FREE(rnb, oa_bufs * sizeof(struct niobuf_remote));
 846         class_export_put(exp);
 847         RETURN(ret);
 848 }