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