lustre/obdfilter/filter.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  linux/fs/filter/filter.c
   5  *
   6  * Copyright (C) 2001  Cluster File Systems, Inc.
   7  *
   8  * This code is issued under the GNU General Public License.
   9  * See the file COPYING in this distribution
  10  *
  11  * by Peter Braam <braam@clusterfs.com>
  12  */
  13
  14 #define EXPORT_SYMTAB
  15 #define DEBUG_SUBSYSTEM S_FILTER
  16
  17 #include <linux/module.h>
  18 #include <linux/obd_filter.h>
  19 #include <linux/ext3_jbd.h>
  20 #include <linux/quotaops.h>
  21
  22 extern struct obd_device obd_dev[MAX_OBD_DEVICES];
  23 long filter_memory;
  24
  25 #define FILTER_ROOTINO 2
  26
  27 #define S_SHIFT 12
  28 static char *obd_type_by_mode[S_IFMT >> S_SHIFT] = {
  29         [0]                     NULL,
  30         [S_IFREG >> S_SHIFT]    "R",
  31         [S_IFDIR >> S_SHIFT]    "D",
  32         [S_IFCHR >> S_SHIFT]    "C",
  33         [S_IFBLK >> S_SHIFT]    "B",
  34         [S_IFIFO >> S_SHIFT]    "F",
  35         [S_IFSOCK >> S_SHIFT]   "S",
  36         [S_IFLNK >> S_SHIFT]    "L"
  37 };
  38
  39 static inline const char *obd_mode_to_type(int mode)
  40 {
  41         return obd_type_by_mode[(mode & S_IFMT) >> S_SHIFT];
  42 }
  43
  44 /* write the pathname into the string */
  45 static int filter_id(char *buf, obd_id id, obd_mode mode)
  46 {
  47         return sprintf(buf, "O/%s/%Ld", obd_mode_to_type(mode),
  48                        (unsigned long long)id);
  49 }
  50
  51 /* setup the object store with correct subdirectories */
  52 static int filter_prep(struct obd_device *obddev)
  53 {
  54         struct obd_run_ctxt saved;
  55         struct filter_obd *filter = &obddev->u.filter;
  56         struct dentry *dentry;
  57         struct file *file;
  58         struct inode *inode;
  59         loff_t off;
  60         int rc = 0;
  61         char rootid[128];
  62         __u64 lastino = 2;
  63         int mode = 0;
  64
  65         push_ctxt(&saved, &filter->fo_ctxt);
  66         dentry = simple_mkdir(current->fs->pwd, "O", 0700);
  67         CDEBUG(D_INODE, "got/created O: %p\n", dentry);
  68         if (IS_ERR(dentry)) {
  69                 rc = PTR_ERR(dentry);
  70                 CERROR("cannot open/create O: rc = %d\n", rc);
  71                 GOTO(out, rc);
  72         }
  73         filter->fo_dentry_O = dentry;
  74         dentry = simple_mkdir(current->fs->pwd, "P", 0700);
  75         CDEBUG(D_INODE, "got/created P: %p\n", dentry);
  76         if (IS_ERR(dentry)) {
  77                 rc = PTR_ERR(dentry);
  78                 CERROR("cannot open/create P: rc = %d\n", rc);
  79                 GOTO(out_O, rc);
  80         }
  81         CDEBUG(D_INODE, "putting P: %p, count = %d\n", dentry,
  82                atomic_read(&dentry->d_count) - 1);
  83         dput(dentry);
  84         dentry = simple_mkdir(current->fs->pwd, "D", 0700);
  85         CDEBUG(D_INODE, "got/created D: %p\n", dentry);
  86         if (IS_ERR(dentry)) {
  87                 rc = PTR_ERR(dentry);
  88                 CERROR("cannot open/create D: rc = %d\n", rc);
  89                 GOTO(out_O, rc);
  90         }
  91         CDEBUG(D_INODE, "putting D: %p, count = %d\n", dentry,
  92                atomic_read(&dentry->d_count) - 1);
  93         dput(dentry);
  94
  95         /*
  96          * Create directories and/or get dentries for each object type.
  97          * This saves us from having to do multiple lookups for each one.
  98          */
  99         for (mode = 0; mode < (S_IFMT >> S_SHIFT); mode++) {
 100                 char *type = obd_type_by_mode[mode];
 101
 102                 if (!type) {
 103                         filter->fo_dentry_O_mode[mode] = NULL;
 104                         continue;
 105                 }
 106                 dentry = simple_mkdir(filter->fo_dentry_O, type, 0700);
 107                 CDEBUG(D_INODE, "got/created O/%s: %p\n", type, dentry);
 108                 if (IS_ERR(dentry)) {
 109                         rc = PTR_ERR(dentry);
 110                         CERROR("cannot create O/%s: rc = %d\n", type, rc);
 111                         GOTO(out_O_mode, rc);
 112                 }
 113                 filter->fo_dentry_O_mode[mode] = dentry;
 114         }
 115
 116         filter_id(rootid, FILTER_ROOTINO, S_IFDIR);
 117         file = filp_open(rootid, O_RDWR | O_CREAT, 0755);
 118         if (IS_ERR(file)) {
 119                 rc = PTR_ERR(file);
 120                 CERROR("OBD filter: cannot open/create root %s: rc = %d\n",
 121                        rootid, rc);
 122                 GOTO(out_O_mode, rc);
 123         }
 124         filp_close(file, 0);
 125
 126         file = filp_open("D/status", O_RDWR | O_CREAT, 0700);
 127         if ( !file || IS_ERR(file) ) {
 128                 rc = PTR_ERR(file);
 129                 CERROR("OBD filter: cannot open/create status %s: rc = %d\n",
 130                        "D/status", rc);
 131                 GOTO(out_O_mode, rc);
 132         }
 133
 134         /* steal operations */
 135         inode = file->f_dentry->d_inode;
 136         filter->fo_fop = file->f_op;
 137         filter->fo_iop = inode->i_op;
 138         filter->fo_aops = inode->i_mapping->a_ops;
 139
 140         off = 0;
 141         if (inode->i_size == 0) {
 142                 ssize_t retval = file->f_op->write(file, (char *)&lastino,
 143                                                    sizeof(lastino), &off);
 144                 if (retval != sizeof(lastino)) {
 145                         CDEBUG(D_INODE, "OBD filter: error writing lastino\n");
 146                         filp_close(file, 0);
 147                         GOTO(out_O_mode, rc = -EIO);
 148                 }
 149         } else {
 150                 ssize_t retval = file->f_op->read(file, (char *)&lastino,
 151                                                   sizeof(lastino), &off);
 152                 if (retval != sizeof(lastino)) {
 153                         CDEBUG(D_INODE, "OBD filter: error reading lastino\n");
 154                         filp_close(file, 0);
 155                         GOTO(out_O_mode, rc = -EIO);
 156                 }
 157         }
 158         filter->fo_lastino = lastino;
 159         filp_close(file, 0);
 160
 161         rc = 0;
 162  out:
 163         pop_ctxt(&saved);
 164
 165         return(rc);
 166
 167 out_O_mode:
 168         while (--mode >= 0) {
 169                 struct dentry *dentry = filter->fo_dentry_O_mode[mode];
 170                 if (dentry) {
 171                         CDEBUG(D_INODE, "putting O/%s: %p, count = %d\n",
 172                                obd_type_by_mode[mode], dentry,
 173                                atomic_read(&dentry->d_count) - 1);
 174                         dput(dentry);
 175                         filter->fo_dentry_O_mode[mode] = NULL;
 176                 }
 177         }
 178 out_O:
 179         CDEBUG(D_INODE, "putting O: %p, count = %d\n", filter->fo_dentry_O,
 180                atomic_read(&filter->fo_dentry_O->d_count) - 1);
 181         dput(filter->fo_dentry_O);
 182         filter->fo_dentry_O = NULL;
 183         goto out;
 184 }
 185
 186 /* cleanup the filter: write last used object id to status file */
 187 static void filter_post(struct obd_device *obddev)
 188 {
 189         struct obd_run_ctxt saved;
 190         struct filter_obd *filter = &obddev->u.filter;
 191         long rc;
 192         struct file *file;
 193         loff_t off = 0;
 194         int mode;
 195
 196         push_ctxt(&saved, &filter->fo_ctxt);
 197         file = filp_open("D/status", O_RDWR | O_CREAT, 0700);
 198         if (IS_ERR(file)) {
 199                 CERROR("OBD filter: cannot create status file\n");
 200                 goto out;
 201         }
 202         rc = file->f_op->write(file, (char *)&filter->fo_lastino,
 203                        sizeof(filter->fo_lastino), &off);
 204         if (rc != sizeof(filter->fo_lastino))
 205                 CERROR("OBD filter: error writing lastino: rc = %ld\n", rc);
 206
 207         rc = filp_close(file, NULL);
 208         if (rc)
 209                 CERROR("OBD filter: cannot close status file: rc = %ld\n", rc);
 210
 211         for (mode = 0; mode < (S_IFMT >> S_SHIFT); mode++) {
 212                 struct dentry *dentry = filter->fo_dentry_O_mode[mode];
 213                 if (dentry) {
 214                         CDEBUG(D_INODE, "putting O/%s: %p, count = %d\n",
 215                                obd_type_by_mode[mode], dentry,
 216                                atomic_read(&dentry->d_count) - 1);
 217                         dput(dentry);
 218                         filter->fo_dentry_O_mode[mode] = NULL;
 219                 }
 220         }
 221         CDEBUG(D_INODE, "putting O: %p, count = %d\n", filter->fo_dentry_O,
 222                atomic_read(&filter->fo_dentry_O->d_count) - 1);
 223         dput(filter->fo_dentry_O);
 224 out:
 225         pop_ctxt(&saved);
 226 }
 227
 228
 229 static __u64 filter_next_id(struct obd_device *obddev)
 230 {
 231         __u64 id;
 232         spin_lock(&obddev->u.filter.fo_lock);
 233         obddev->u.filter.fo_lastino++;
 234         id =    obddev->u.filter.fo_lastino;
 235         spin_unlock(&obddev->u.filter.fo_lock);
 236         return id;
 237 }
 238
 239 /* how to get files, dentries, inodes from object id's */
 240 /* parent i_sem is already held if needed for exclusivity */
 241 static struct dentry *filter_fid2dentry(struct obd_device *obddev,
 242                                         struct dentry *dparent,
 243                                         __u64 id, __u32 type)
 244 {
 245         struct super_block *sb = obddev->u.filter.fo_sb;
 246         struct dentry *dchild;
 247         char name[32];
 248         int len;
 249         ENTRY;
 250
 251         if (!sb || !sb->s_dev) {
 252                 CERROR("fatal: device not initialized.\n");
 253                 RETURN(ERR_PTR(-ENXIO));
 254         }
 255
 256         if (id == 0) {
 257                 CERROR("fatal: invalid object #0\n");
 258                 RETURN(ERR_PTR(-ESTALE));
 259         }
 260
 261         if (!(type & S_IFMT)) {
 262                 CERROR("OBD %s, object %Lu has bad type: %o\n", __FUNCTION__,
 263                        (unsigned long long)id, type);
 264                 RETURN(ERR_PTR(-EINVAL));
 265         }
 266
 267         len = sprintf(name, "%Ld", id);
 268         CDEBUG(D_INODE, "opening object O/%s/%s\n", obd_mode_to_type(type),
 269                name);
 270         dchild = lookup_one_len(name, dparent, len);
 271         CDEBUG(D_INODE, "got child obj O/%s/%s: %p, count = %d\n",
 272                obd_mode_to_type(type), name, dchild,
 273                atomic_read(&dchild->d_count));
 274
 275         if (IS_ERR(dchild)) {
 276                 CERROR("child lookup error %ld\n", PTR_ERR(dchild));
 277                 RETURN(dchild);
 278         }
 279
 280         RETURN(dchild);
 281 }
 282
 283 static struct file *filter_obj_open(struct obd_device *obddev,
 284                                     __u64 id, __u32 type)
 285 {
 286         struct super_block *sb = obddev->u.filter.fo_sb;
 287         struct obd_run_ctxt saved;
 288         char name[24];
 289         struct file *file;
 290         ENTRY;
 291
 292         if (!sb || !sb->s_dev) {
 293                 CERROR("fatal: device not initialized.\n");
 294                 RETURN(ERR_PTR(-ENXIO));
 295         }
 296
 297         if (!id) {
 298                 CERROR("fatal: invalid obdo %Lu\n", (unsigned long long)id);
 299                 RETURN(ERR_PTR(-ESTALE));
 300         }
 301
 302         if (!(type & S_IFMT)) {
 303                 CERROR("OBD %s, no type (%Ld), mode %o!\n", __FUNCTION__,
 304                        (unsigned long long)id, type);
 305                 RETURN(ERR_PTR(-EINVAL));
 306         }
 307
 308         filter_id(name, id, type);
 309         push_ctxt(&saved, &obddev->u.filter.fo_ctxt);
 310         file = filp_open(name, O_RDONLY | O_LARGEFILE, 0 /* type? */);
 311         pop_ctxt(&saved);
 312
 313         CDEBUG(D_INODE, "opening obdo %s: rc = %p\n", name, file);
 314
 315         if (IS_ERR(file))
 316                 file = NULL;
 317         RETURN(file);
 318 }
 319
 320 static struct dentry *filter_parent(struct obd_device *obddev, obd_mode mode)
 321 {
 322         struct filter_obd *filter = &obddev->u.filter;
 323
 324         return filter->fo_dentry_O_mode[(mode & S_IFMT) >> S_SHIFT];
 325 }
 326
 327
 328 static struct inode *filter_inode_from_obj(struct obd_device *obddev,
 329                                            __u64 id, __u32 type)
 330 {
 331         struct dentry *dentry;
 332         struct inode *inode;
 333
 334         dentry = filter_fid2dentry(obddev, filter_parent(obddev, type),
 335                                    id, type);
 336         if (IS_ERR(dentry)) {
 337                 CERROR("%s: lookup failed: rc = %ld\n", __FUNCTION__,
 338                        PTR_ERR(dentry));
 339                 RETURN(NULL);
 340         }
 341
 342         lock_kernel();
 343         inode = iget(dentry->d_inode->i_sb, dentry->d_inode->i_ino);
 344         unlock_kernel();
 345         CDEBUG(D_INODE, "put child %p, count = %d\n", dentry,
 346                atomic_read(&dentry->d_count) - 1);
 347         dput(dentry);
 348         CDEBUG(D_INODE, "got inode %p (%ld), count = %d\n", inode, inode->i_ino,
 349                atomic_read(&inode->i_count));
 350         return inode;
 351 }
 352
 353 /* obd methods */
 354 static int filter_connect(struct obd_conn *conn)
 355 {
 356         int rc;
 357
 358         MOD_INC_USE_COUNT;
 359         rc = gen_connect(conn);
 360
 361         if (rc)
 362                 MOD_DEC_USE_COUNT;
 363
 364         return rc;
 365 }
 366
 367 static int filter_disconnect(struct obd_conn *conn)
 368 {
 369         int rc;
 370
 371         rc = gen_disconnect(conn);
 372         if (!rc)
 373                 MOD_DEC_USE_COUNT;
 374
 375         /* XXX cleanup preallocated inodes */
 376         return rc;
 377 }
 378
 379 /* mount the file system (secretly) */
 380 static int filter_setup(struct obd_device *obddev, obd_count len, void *buf)
 381 {
 382         struct obd_ioctl_data* data = buf;
 383         struct filter_obd *filter;
 384         struct vfsmount *mnt;
 385         int err = 0;
 386         ENTRY;
 387
 388         if (!data->ioc_inlbuf1 || !data->ioc_inlbuf2)
 389                 RETURN(-EINVAL);
 390
 391         MOD_INC_USE_COUNT;
 392         mnt = do_kern_mount(data->ioc_inlbuf2, 0, data->ioc_inlbuf1, NULL);
 393         err = PTR_ERR(mnt);
 394         if (IS_ERR(mnt))
 395                 GOTO(err_dec, err);
 396
 397         filter = &obddev->u.filter;;
 398         filter->fo_sb = mnt->mnt_root->d_inode->i_sb;
 399         /* XXX is this even possible if do_kern_mount succeeded? */
 400         if (!filter->fo_sb)
 401                 GOTO(err_put, err = -ENODEV);
 402
 403         filter->fo_vfsmnt = mnt;
 404         filter->fo_fstype = strdup(data->ioc_inlbuf2);
 405
 406         filter->fo_ctxt.pwdmnt = mnt;
 407         filter->fo_ctxt.pwd = mnt->mnt_root;
 408         filter->fo_ctxt.fs = KERNEL_DS;
 409
 410         err = filter_prep(obddev);
 411         if (err)
 412                 GOTO(err_kfree, err);
 413         spin_lock_init(&filter->fo_lock);
 414
 415         RETURN(0);
 416
 417 err_kfree:
 418         kfree(filter->fo_fstype);
 419 err_put:
 420         unlock_kernel();
 421         mntput(filter->fo_vfsmnt);
 422         filter->fo_sb = 0;
 423         lock_kernel();
 424
 425 err_dec:
 426         MOD_DEC_USE_COUNT;
 427         return err;
 428 }
 429
 430
 431 static int filter_cleanup(struct obd_device * obddev)
 432 {
 433         struct super_block *sb;
 434         ENTRY;
 435
 436         if (!(obddev->obd_flags & OBD_SET_UP))
 437                 RETURN(0);
 438
 439         if (!list_empty(&obddev->obd_gen_clients)) {
 440                 CERROR("still has clients!\n");
 441                 RETURN(-EBUSY);
 442         }
 443
 444         sb = obddev->u.filter.fo_sb;
 445         if (!obddev->u.filter.fo_sb)
 446                 RETURN(0);
 447
 448         filter_post(obddev);
 449
 450         shrink_dcache_parent(sb->s_root);
 451         unlock_kernel();
 452         mntput(obddev->u.filter.fo_vfsmnt);
 453         obddev->u.filter.fo_sb = 0;
 454         kfree(obddev->u.filter.fo_fstype);
 455
 456         lock_kernel();
 457
 458         MOD_DEC_USE_COUNT;
 459         RETURN(0);
 460 }
 461
 462
 463 static inline void filter_from_inode(struct obdo *oa, struct inode *inode)
 464 {
 465         int type = oa->o_mode & S_IFMT;
 466         ENTRY;
 467
 468         CDEBUG(D_INFO, "src inode %ld, dst obdo %ld valid 0x%08x\n",
 469                inode->i_ino, (long)oa->o_id, oa->o_valid);
 470         obdo_from_inode(oa, inode);
 471         oa->o_mode &= ~S_IFMT;
 472         oa->o_mode |= type;
 473
 474         if (S_ISCHR(inode->i_mode) || S_ISBLK(inode->i_mode)) {
 475                 obd_rdev rdev = kdev_t_to_nr(inode->i_rdev);
 476                 CDEBUG(D_INODE, "copying device %x from inode to obdo\n",
 477                        rdev);
 478                 *((obd_rdev *)oa->o_inline) = rdev;
 479                 oa->o_obdflags |= OBD_FL_INLINEDATA;
 480                 oa->o_valid |= OBD_MD_FLINLINE;
 481         }
 482
 483 #if 0
 484         else if (filter_has_inline(inode)) {
 485                 CDEBUG(D_INFO, "copying inline from inode to obdo\n");
 486                 memcpy(oa->o_inline, inode->u.ext2_i.i_data,
 487                        MIN(sizeof(inode->u.ext2_i.i_data),OBD_INLINESZ));
 488                 oa->o_obdflags |= OBD_FL_INLINEDATA;
 489                 oa->o_valid |= OBD_MD_FLINLINE;
 490         }
 491
 492         if (filter_has_obdmd(inode)) {
 493                 /* XXX this will change when we don't store the obdmd in data */
 494                 CDEBUG(D_INFO, "copying obdmd from inode to obdo\n");
 495                 memcpy(oa->o_obdmd, inode->u.ext2_i.i_data,
 496                        MIN(sizeof(inode->u.ext2_i.i_data),OBD_INLINESZ));
 497                 oa->o_obdflags |= OBD_FL_OBDMDEXISTS;
 498                 oa->o_valid |= OBD_MD_FLOBDMD;
 499         }
 500 #endif
 501         EXIT;
 502 }
 503
 504 static int filter_getattr(struct obd_conn *conn, struct obdo *oa)
 505 {
 506         struct obd_device *obddev;
 507         struct dentry *dentry;
 508         ENTRY;
 509
 510         if (!gen_client(conn)) {
 511                 CDEBUG(D_IOCTL, "fatal: invalid client %u\n", conn->oc_id);
 512                 RETURN(-EINVAL);
 513         }
 514
 515         obddev = conn->oc_dev;
 516         dentry = filter_fid2dentry(obddev, filter_parent(obddev, oa->o_mode),
 517                                    oa->o_id, oa->o_mode);
 518         if (IS_ERR(dentry))
 519                 RETURN(PTR_ERR(dentry));
 520
 521         oa->o_valid &= ~OBD_MD_FLID;
 522         filter_from_inode(oa, dentry->d_inode);
 523
 524         dput(dentry);
 525         RETURN(0);
 526 }
 527
 528 static int filter_setattr(struct obd_conn *conn, struct obdo *oa)
 529 {
 530         struct obd_run_ctxt saved;
 531         struct obd_device *obddev;
 532         struct dentry *dentry;
 533         struct iattr iattr;
 534         struct inode *inode;
 535         int rc;
 536         ENTRY;
 537
 538         if (!gen_client(conn)) {
 539                 CDEBUG(D_IOCTL, "invalid client %u\n", conn->oc_id);
 540                 RETURN(-EINVAL);
 541         }
 542
 543         obddev = conn->oc_dev;
 544         dentry = filter_fid2dentry(obddev, filter_parent(obddev, oa->o_mode),
 545                                    oa->o_id, oa->o_mode);
 546         if (IS_ERR(dentry))
 547                 RETURN(PTR_ERR(dentry));
 548
 549         inode = dentry->d_inode;
 550         iattr_from_obdo(&iattr, oa);
 551         iattr.ia_mode &= ~S_IFMT;
 552         iattr.ia_mode |= S_IFREG;
 553         lock_kernel();
 554         if (iattr.ia_mode & ATTR_SIZE)
 555                 down(&inode->i_sem);
 556         push_ctxt(&saved, &conn->oc_dev->u.filter.fo_ctxt);
 557         if (inode->i_op->setattr)
 558                 rc = inode->i_op->setattr(dentry, &iattr);
 559         else
 560                 rc = inode_setattr(inode, &iattr);
 561         pop_ctxt(&saved);
 562         if (iattr.ia_mode & ATTR_SIZE)
 563                 up(&inode->i_sem);
 564         unlock_kernel();
 565
 566         CDEBUG(D_INODE, "put dentry %p, count = %d\n", inode,
 567                atomic_read(&dentry->d_count) - 1);
 568         dput(dentry);
 569         RETURN(rc);
 570 }
 571
 572 static int filter_open(struct obd_conn *conn, struct obdo *oa)
 573 {
 574         struct obd_device *obddev;
 575         struct dentry *dentry;
 576         /* ENTRY; */
 577
 578         if (!gen_client(conn)) {
 579                 CDEBUG(D_IOCTL, "fatal: invalid client %u\n", conn->oc_id);
 580                 RETURN(-EINVAL);
 581         }
 582
 583         obddev = conn->oc_dev;
 584         dentry = filter_fid2dentry(obddev, filter_parent(obddev, oa->o_mode),
 585                                    oa->o_id, oa->o_mode);
 586         if (IS_ERR(dentry))
 587                 RETURN(PTR_ERR(dentry));
 588
 589         return 0;
 590 } /* filter_open */
 591
 592 static int filter_close(struct obd_conn *conn, struct obdo *oa)
 593 {
 594         struct obd_device *obddev;
 595         struct dentry *dentry;
 596         /* ENTRY; */
 597
 598         if (!gen_client(conn)) {
 599                 CDEBUG(D_IOCTL, "fatal: invalid client %u\n", conn->oc_id);
 600                 RETURN(-EINVAL);
 601         }
 602
 603         obddev = conn->oc_dev;
 604         dentry = filter_fid2dentry(obddev, filter_parent(obddev, oa->o_mode),
 605                                    oa->o_id, oa->o_mode);
 606         if (IS_ERR(dentry))
 607                 RETURN(PTR_ERR(dentry));
 608
 609         CDEBUG(D_INODE, "put dentry %p, count = %d\n", dentry,
 610                atomic_read(&dentry->d_count) - 1);
 611         dput(dentry);  /* for the close */
 612         CDEBUG(D_INODE, "put dentry %p, count = %d\n", dentry,
 613                atomic_read(&dentry->d_count) - 1);
 614         dput(dentry);  /* for this call */
 615         return 0;
 616 } /* filter_close */
 617
 618 static int filter_create(struct obd_conn* conn, struct obdo *oa)
 619 {
 620         char name[64];
 621         struct obd_run_ctxt saved;
 622         struct file *file;
 623         int mode;
 624         struct obd_device *obddev = conn->oc_dev;
 625         struct iattr;
 626         ENTRY;
 627
 628         if (!gen_client(conn)) {
 629                 CERROR("invalid client %u\n", conn->oc_id);
 630                 return -EINVAL;
 631         }
 632
 633         oa->o_id = filter_next_id(conn->oc_dev);
 634         if (!(oa->o_mode && S_IFMT)) {
 635                 CERROR("filter obd: no type!\n");
 636                 return -ENOENT;
 637         }
 638
 639         filter_id(name, oa->o_id, oa->o_mode);
 640         mode = (oa->o_mode & ~S_IFMT) | S_IFREG;
 641         push_ctxt(&saved, &obddev->u.filter.fo_ctxt);
 642         file = filp_open(name, O_RDONLY | O_CREAT, mode);
 643         pop_ctxt(&saved);
 644         if (IS_ERR(file)) {
 645                 CERROR("Error mknod obj %s, err %ld\n", name, PTR_ERR(file));
 646                 return -ENOENT;
 647         }
 648         filp_close(file, 0);
 649
 650         /* Set flags for fields we have set in the inode struct */
 651         oa->o_valid |= OBD_MD_FLID | OBD_MD_FLBLKSZ | OBD_MD_FLBLOCKS |
 652                  OBD_MD_FLMTIME | OBD_MD_FLATIME | OBD_MD_FLCTIME |
 653                  OBD_MD_FLUID | OBD_MD_FLGID;
 654
 655         /* XXX Hmm, shouldn't we copy the fields into the obdo here? */
 656         return 0;
 657 }
 658
 659 static int filter_destroy(struct obd_conn *conn, struct obdo *oa)
 660 {
 661         struct obd_run_ctxt saved;
 662         struct obd_device *obddev;
 663         struct obd_client *cli;
 664         struct inode *inode;
 665         struct dentry *dir_dentry, *object_dentry;
 666         int rc;
 667         ENTRY;
 668
 669         if (!(cli = gen_client(conn))) {
 670                 CERROR("invalid client %u\n", conn->oc_id);
 671                 RETURN(-EINVAL);
 672         }
 673
 674         CDEBUG(D_INODE, "destroying object %Ld\n",oa->o_id);
 675         obddev = conn->oc_dev;
 676
 677         dir_dentry = filter_parent(obddev, oa->o_mode);
 678         down(&dir_dentry->d_inode->i_sem);
 679
 680         object_dentry = filter_fid2dentry(obddev, dir_dentry, oa->o_id,
 681                                           oa->o_mode);
 682         if (IS_ERR(object_dentry))
 683                 GOTO(out, rc = -ENOENT);
 684
 685         inode = object_dentry->d_inode;
 686         if (inode->i_nlink != 1) {
 687                 CERROR("destroying inode with nlink = %d\n", inode->i_nlink);
 688                 inode->i_nlink = 1;
 689         }
 690         inode->i_mode = S_IFREG;
 691
 692         push_ctxt(&saved, &obddev->u.filter.fo_ctxt);
 693         rc = vfs_unlink(dir_dentry->d_inode, object_dentry);
 694         pop_ctxt(&saved);
 695         CDEBUG(D_INODE, "put child %p, count = %d\n", object_dentry,
 696                atomic_read(&object_dentry->d_count) - 1);
 697         dput(object_dentry);
 698
 699         EXIT;
 700 out:
 701         up(&dir_dentry->d_inode->i_sem);
 702         return rc;
 703 }
 704
 705 /* NB count and offset are used for punch, but not truncate */
 706 static int filter_truncate(struct obd_conn *conn, struct obdo *oa,
 707                            obd_size count, obd_off offset)
 708 {
 709         int error;
 710         ENTRY;
 711
 712         CDEBUG(D_INODE, "calling truncate for object #%Ld, valid = %x, "
 713                "o_size = %Ld\n", oa->o_id, oa->o_valid, oa->o_size);
 714         error = filter_setattr(conn, oa);
 715         oa->o_valid = OBD_MD_FLBLOCKS | OBD_MD_FLCTIME | OBD_MD_FLMTIME;
 716
 717         RETURN(error);
 718 }
 719
 720 /* buffer must lie in user memory here */
 721 static int filter_read(struct obd_conn *conn, struct obdo *oa, char *buf,
 722                         obd_size *count, obd_off offset)
 723 {
 724         struct file * file;
 725         unsigned long retval;
 726         int err;
 727         ENTRY;
 728
 729         if (!gen_client(conn)) {
 730                 CDEBUG(D_IOCTL, "invalid client %u\n", conn->oc_id);
 731                 RETURN(-EINVAL);
 732         }
 733
 734         file = filter_obj_open(conn->oc_dev, oa->o_id, oa->o_mode);
 735         if (IS_ERR(file))
 736                 RETURN(PTR_ERR(file));
 737
 738         /* count doubles as retval */
 739         retval = file->f_op->read(file, buf, *count, (loff_t *)&offset);
 740         filp_close(file, 0);
 741
 742         if ( retval >= 0 ) {
 743                 err = 0;
 744                 *count = retval;
 745         } else {
 746                 err = retval;
 747                 *count = 0;
 748         }
 749
 750         return err;
 751 }
 752
 753
 754 /* buffer must lie in user memory here */
 755 static int filter_write(struct obd_conn *conn, struct obdo *oa, char *buf,
 756                          obd_size *count, obd_off offset)
 757 {
 758         struct obd_run_ctxt saved;
 759         int err;
 760         struct file * file;
 761         unsigned long retval;
 762         ENTRY;
 763
 764         if (!gen_client(conn)) {
 765                 CDEBUG(D_IOCTL, "invalid client %u\n", conn->oc_id);
 766                 RETURN(-EINVAL);
 767         }
 768
 769         file = filter_obj_open(conn->oc_dev, oa->o_id, oa->o_mode);
 770         if (IS_ERR(file))
 771                 RETURN(PTR_ERR(file));
 772
 773         /* count doubles as retval */
 774         push_ctxt(&saved, &conn->oc_dev->u.filter.fo_ctxt);
 775         retval = file->f_op->write(file, buf, *count, (loff_t *)&offset);
 776         pop_ctxt(&saved);
 777         filp_close(file, 0);
 778
 779         if ( retval >= 0 ) {
 780                 err = 0;
 781                 *count = retval;
 782                 EXIT;
 783         } else {
 784                 err = retval;
 785                 *count = 0;
 786                 EXIT;
 787         }
 788
 789         return err;
 790 } /* filter_write */
 791
 792 static int filter_pgcache_brw(int rw, struct obd_conn *conn, obd_count num_oa,
 793                                struct obdo **oa, obd_count *oa_bufs,
 794                                struct page **pages, obd_size *count,
 795                                obd_off *offset, obd_flag *flags)
 796 {
 797         struct obd_run_ctxt      saved;
 798         struct super_block      *sb;
 799         int                      onum;          /* index to oas */
 800         int                      pnum;          /* index to pages (bufs) */
 801         unsigned long            retval;
 802         int                      error;
 803         struct file             *file;
 804         ENTRY;
 805
 806         if (!gen_client(conn)) {
 807                 CDEBUG(D_IOCTL, "invalid client %u\n", conn->oc_id);
 808                 RETURN(-EINVAL);
 809         }
 810
 811         sb = conn->oc_dev->u.filter.fo_sb;
 812         // if (rw == WRITE)
 813         push_ctxt(&saved, &conn->oc_dev->u.filter.fo_ctxt);
 814         pnum = 0; /* pnum indexes buf 0..num_pages */
 815         for (onum = 0; onum < num_oa; onum++) {
 816                 int pg;
 817
 818                 file = filter_obj_open(conn->oc_dev, oa[onum]->o_id,
 819                                        oa[onum]->o_mode);
 820                 if (IS_ERR(file))
 821                         GOTO(out, retval = PTR_ERR(file));
 822
 823                 /* count doubles as retval */
 824                 for (pg = 0; pg < oa_bufs[onum]; pg++) {
 825                         CDEBUG(D_INODE, "OP %d obdo no/pno: (%d,%d) (%ld,%ld) "
 826                                "off count (%Ld,%Ld)\n",
 827                                rw, onum, pnum, file->f_dentry->d_inode->i_ino,
 828                                (unsigned long)offset[pnum] >> PAGE_CACHE_SHIFT,
 829                                (unsigned long long)offset[pnum],
 830                                (unsigned long long)count[pnum]);
 831                         if (rw == WRITE) {
 832                                 loff_t off;
 833                                 char *buffer;
 834                                 off = offset[pnum];
 835                                 buffer = kmap(pages[pnum]);
 836                                 retval = file->f_op->write(file, buffer, count[pnum], &off);
 837                                 kunmap(pages[pnum]);
 838                                 CDEBUG(D_INODE, "retval %ld\n", retval);
 839                         } else {
 840                                 loff_t off = offset[pnum];
 841                                 char *buffer = kmap(pages[pnum]);
 842
 843                                 if (off >= file->f_dentry->d_inode->i_size) {
 844                                         memset(buffer, 0, count[pnum]);
 845                                         retval = count[pnum];
 846                                 } else {
 847                                         retval = file->f_op->read(file, buffer, count[pnum], &off);
 848                                 }
 849                                 kunmap(pages[pnum]);
 850
 851                                 if (retval != count[pnum]) {
 852                                         filp_close(file, 0);
 853                                         GOTO(out, retval = -EIO);
 854                                 }
 855                                 CDEBUG(D_INODE, "retval %ld\n", retval);
 856                         }
 857                         pnum++;
 858                 }
 859                 /* sizes and blocks are set by generic_file_write */
 860                 /* ctimes/mtimes will follow with a setattr call */
 861                 filp_close(file, 0);
 862         }
 863
 864         EXIT;
 865 out:
 866         // if (rw == WRITE)
 867         pop_ctxt(&saved);
 868         error = (retval >= 0) ? 0 : retval;
 869         return error;
 870 }
 871
 872
 873 struct inode *ioobj_to_inode(struct obd_conn *conn, struct obd_ioobj *o)
 874 {
 875         struct super_block *sb = conn->oc_dev->u.filter.fo_sb;
 876         struct inode *inode = NULL;
 877         ENTRY;
 878
 879         if (!sb || !sb->s_dev) {
 880                 CDEBUG(D_SUPER, "fatal: device not initialized.\n");
 881                 RETURN(NULL);
 882         }
 883
 884         if ( !o->ioo_id ) {
 885                 CDEBUG(D_INODE, "fatal: invalid obdo %lu\n", (long)o->ioo_id);
 886                 RETURN(NULL);
 887         }
 888
 889         inode = filter_inode_from_obj(conn->oc_dev, o->ioo_id, S_IFREG);
 890         if (!inode || inode->i_nlink == 0 || is_bad_inode(inode)) {
 891                 CERROR("from obdo - fatal: invalid inode %ld (%s).\n",
 892                        (long)o->ioo_id, inode ? inode->i_nlink ? "bad inode" :
 893                        "no links" : "NULL");
 894                 iput(inode);
 895                 RETURN(NULL);
 896         }
 897
 898         RETURN(inode);
 899 }
 900
 901 /*
 902  * Calculate the number of buffer credits needed to write multiple pages in
 903  * a single ext3/extN transaction.  No, this shouldn't be here, but as yet
 904  * ext3 doesn't have a nice API for calculating this sort of thing in advance.
 905  *
 906  * See comment above ext3_writepage_trans_blocks for details.  We assume
 907  * no data journaling is being done, but it does allow for all of the pages
 908  * being non-contiguous.  If we are guaranteed contiguous pages we could
 909  * reduce the number of (d)indirect blocks a lot.
 910  *
 911  * With N blocks per page and P pages, for each inode we have at most:
 912  * N*P indirect
 913  * min(N*P, blocksize/4 + 1) dindirect blocks
 914  * 1 tindirect
 915  *
 916  * For the entire filesystem, we have at most:
 917  * min(sum(nindir + P), ngroups) bitmap blocks (from the above)
 918  * min(sum(nindir + P), gdblocks) group descriptor blocks (from the above)
 919  * 1 inode block
 920  * 1 superblock
 921  * 2 * EXT3_SINGLEDATA_TRANS_BLOCKS for the quota files
 922  */
 923 static int ext3_credits_needed(struct super_block *sb, int objcount,
 924                                struct obd_ioobj *obj)
 925 {
 926         struct obd_ioobj *o = obj;
 927         int blockpp = 1 << (PAGE_CACHE_SHIFT - sb->s_blocksize_bits);
 928         int addrpp = EXT3_ADDR_PER_BLOCK(sb) * blockpp;
 929         int nbitmaps = 0;
 930         int ngdblocks = 0;
 931         int needed = objcount + 1;
 932         int i;
 933
 934         for (i = 0; i < objcount; i++, o++) {
 935                 int nblocks = o->ioo_bufcnt * blockpp;
 936                 int ndindirect = min(nblocks, addrpp + 1);
 937                 int nindir = nblocks + ndindirect + 1;
 938
 939                 nbitmaps += nindir + nblocks;
 940                 ngdblocks += nindir + nblocks;
 941
 942                 needed += nindir;
 943         }
 944
 945         if (nbitmaps > EXT3_SB(sb)->s_groups_count)
 946                 nbitmaps = EXT3_SB(sb)->s_groups_count;
 947         if (ngdblocks > EXT3_SB(sb)->s_gdb_count)
 948                 ngdblocks = EXT3_SB(sb)->s_gdb_count;
 949
 950         needed += nbitmaps + ngdblocks;
 951
 952 #ifdef CONFIG_QUOTA
 953         /* We assume that there will be 1 bit set in s_dquot.flags for each
 954          * quota file that is active.  This is at least true for now.
 955          */
 956         needed += hweight32(sb_any_quota_enabled(sb)) *
 957                 EXT3_SINGLEDATA_TRANS_BLOCKS;
 958 #endif
 959
 960         return needed;
 961 }
 962
 963 /* We have to start a huge journal transaction here to hold all of the
 964  * metadata for the pages being written here.  This is necessitated by
 965  * the fact that we do lots of prepare_write operations before we do
 966  * any of the matching commit_write operations, so even if we split
 967  * up to use "smaller" transactions none of them could complete until
 968  * all of them were opened.  By having a single journal transaction,
 969  * we eliminate duplicate reservations for common blocks like the
 970  * superblock and group descriptors or bitmaps.
 971  *
 972  * We will start the transaction here, but each prepare_write will
 973  * add a refcount to the transaction, and each commit_write will
 974  * remove a refcount.  The transaction will be closed when all of
 975  * the pages have been written.
 976  */
 977 static void *ext3_filter_journal_start(struct filter_obd *filter,
 978                                        int objcount, struct obd_ioobj *obj,
 979                                        int niocount, struct niobuf_remote *nb)
 980 {
 981         journal_t *journal = NULL;
 982         handle_t *handle = NULL;
 983         int needed;
 984
 985         /* Assumes ext3 and extN have same sb_info layout, but avoids issues
 986          * with having extN built properly before filterobd for now.
 987          */
 988         journal = EXT3_SB(filter->fo_sb)->s_journal;
 989         needed = ext3_credits_needed(filter->fo_sb, objcount, obj);
 990
 991         /* The number of blocks we could _possibly_ dirty can very large.
 992          * We reduce our request if it is absurd (and we couldn't get that
 993          * many credits for a single handle anyways).
 994          *
 995          * At some point we have to limit the size of I/Os sent at one time,
 996          * increase the size of the journal, or we have to calculate the
 997          * actual journal requirements more carefully by checking all of
 998          * the blocks instead of being maximally pessimistic.  It remains to
 999          * be seen if this is a real problem or not.
1000          */
1001         if (needed > journal->j_max_transaction_buffers) {
1002                 CERROR("want too many journal credits (%d) using %d instead\n",
1003                        needed, journal->j_max_transaction_buffers);
1004                 needed = journal->j_max_transaction_buffers;
1005         }
1006
1007         handle = journal_start(journal, needed);
1008         if (IS_ERR(handle))
1009                 CERROR("can't get handle for %d credits: rc = %ld\n", needed,
1010                        PTR_ERR(handle));
1011
1012         return(handle);
1013 }
1014
1015 static void *filter_journal_start(void **journal_save,
1016                                   struct filter_obd *filter,
1017                                   int objcount, struct obd_ioobj *obj,
1018                                   int niocount, struct niobuf_remote *nb)
1019 {
1020         void *handle = NULL;
1021
1022         /* This may not be necessary - we probably never have a
1023          * transaction started when we enter here, so we can
1024          * remove the saving of the journal state entirely.
1025          * For now leave it in just to see if it ever happens.
1026          */
1027         *journal_save = current->journal_info;
1028         if (*journal_save) {
1029                 CERROR("Already have handle %p???\n", *journal_save);
1030                 LBUG();
1031                 current->journal_info = NULL;
1032         }
1033
1034         if (!strcmp(filter->fo_fstype, "ext3") ||
1035             !strcmp(filter->fo_fstype, "extN"))
1036                 handle = ext3_filter_journal_start(filter, objcount, obj,
1037                                                    niocount, nb);
1038         return handle;
1039 }
1040
1041 static int ext3_filter_journal_stop(void *handle)
1042 {
1043         int rc;
1044
1045         /* We got a refcount on the handle for each call to prepare_write,
1046          * so we can drop the "parent" handle here to avoid the need for
1047          * osc to call back into filterobd to close the handle.  The
1048          * remaining references will be dropped in commit_write.
1049          */
1050         rc = journal_stop((handle_t *)handle);
1051
1052         return rc;
1053 }
1054
1055 static int filter_journal_stop(void *journal_save, struct filter_obd *filter,
1056                                void *handle)
1057 {
1058         int rc = 0;
1059
1060         if (!strcmp(filter->fo_fstype, "ext3") ||
1061             !strcmp(filter->fo_fstype, "extN"))
1062                 rc = ext3_filter_journal_stop(handle);
1063
1064         current->journal_info = journal_save;
1065
1066         return rc;
1067 }
1068
1069 struct page *filter_get_page_write(struct inode *inode, unsigned long index,
1070                                    struct niobuf_local *lnb)
1071 {
1072         struct address_space *mapping = inode->i_mapping;
1073         struct page *page;
1074         int rc;
1075
1076         //ASSERT_PAGE_INDEX(index, GOTO(err, rc = -EINVAL));
1077         page = grab_cache_page_nowait(mapping, index); /* locked page */
1078
1079         /* This page is currently locked, so we grab a new one temporarily */
1080         if (!page) {
1081                 unsigned long addr;
1082                 addr = __get_free_pages(GFP_KERNEL, 0);
1083                 if (!addr) {
1084                         CERROR("no memory for a temp page\n");
1085                         LBUG();
1086                         GOTO(err, rc = -ENOMEM);
1087                 }
1088                 page = virt_to_page(addr);
1089                 kmap(page);
1090                 page->index = index;
1091                 lnb->flags |= N_LOCAL_TEMP_PAGE;
1092         } else if (!IS_ERR(page)) {
1093                 /* Note: Called with "O" and "PAGE_SIZE" this is essentially
1094                  * a no-op for most filesystems, because we write the whole
1095                  * page.  For partial-page I/O this will read in the page.
1096                  */
1097                 rc = mapping->a_ops->prepare_write(NULL, page, 0, PAGE_SIZE);
1098                 if (rc) {
1099                         CERROR("page index %lu, rc = %d\n", index, rc);
1100                         if (rc != -ENOSPC)
1101                                 LBUG();
1102                         GOTO(err_unlock, rc);
1103                 }
1104                 /* XXX not sure if we need this if we are overwriting page */
1105                 if (PageError(page)) {
1106                         CERROR("error on page index %lu, rc = %d\n", index, rc);
1107                         LBUG();
1108                         GOTO(err_unlock, rc = -EIO);
1109                 }
1110
1111                 kmap(page);
1112         }
1113         return page;
1114
1115 err_unlock:
1116         unlock_page(page);
1117         lustre_put_page(page);
1118 err:
1119         return ERR_PTR(rc);
1120 }
1121
1122 static int filter_preprw(int cmd, struct obd_conn *conn,
1123                          int objcount, struct obd_ioobj *obj,
1124                          int niocount, struct niobuf_remote *nb,
1125                          struct niobuf_local *res, void **desc_private)
1126 {
1127         struct obd_run_ctxt saved;
1128         struct obd_device *obddev;
1129         struct obd_ioobj *o = obj;
1130         struct niobuf_remote *b = nb;
1131         struct niobuf_local *r = res;
1132         void *journal_save = NULL;
1133         int rc = 0;
1134         int i;
1135         ENTRY;
1136
1137         memset(res, 0, sizeof(*res) * niocount);
1138         obddev = conn->oc_dev;
1139
1140         push_ctxt(&saved, &obddev->u.filter.fo_ctxt);
1141
1142         if (cmd == OBD_BRW_WRITE) {
1143                 *desc_private = filter_journal_start(&journal_save,
1144                                                      &obddev->u.filter,
1145                                                      objcount, obj, niocount,
1146                                                      nb);
1147                 if (IS_ERR(*desc_private))
1148                         GOTO(out_ctxt, rc = PTR_ERR(*desc_private));
1149         }
1150
1151         for (i = 0; i < objcount; i++, o++) {
1152                 struct dentry *dentry;
1153                 struct inode *inode;
1154                 int j;
1155
1156                 dentry = filter_fid2dentry(obddev,
1157                                            filter_parent(obddev, S_IFREG),
1158                                            o->ioo_id, S_IFREG);
1159                 inode = dentry->d_inode;
1160
1161                 for (j = 0; j < o->ioo_bufcnt; j++, b++, r++) {
1162                         unsigned long index = b->offset >> PAGE_SHIFT;
1163                         struct page *page;
1164
1165                         /* XXX  We _might_ change this to a dcount if we
1166                          * wanted to pass a dentry pointer in the niobuf
1167                          * to avoid doing so many igets on an inode we
1168                          * already have.  It appears to be solely for the
1169                          * purpose of having a refcount that we can drop
1170                          * in commitrw where we get one call per page.
1171                          */
1172                         if (j > 0)
1173                                 r->dentry = dget(dentry);
1174                         else
1175                                 r->dentry = dentry;
1176
1177                         /* FIXME: we need to iput all inodes on error */
1178                         if (!inode)
1179                                 RETURN(-EINVAL);
1180
1181                         if (cmd == OBD_BRW_WRITE) {
1182                                 page = filter_get_page_write(inode, index, r);
1183
1184                                 /* We unlock the page to avoid deadlocks with
1185                                  * the page I/O because we are preparing
1186                                  * multiple pages at one time and we have lock
1187                                  * ordering problems.  Lustre I/O and disk I/O
1188                                  * on this page can happen concurrently.
1189                                  */
1190                         } else
1191                                 page = lustre_get_page_read(inode, index);
1192
1193                         /* FIXME: we need to clean up here... */
1194                         if (IS_ERR(page))
1195                                 RETURN(PTR_ERR(page));
1196
1197                         r->addr = page_address(page);
1198                         r->offset = b->offset;
1199                         r->page = page;
1200                         r->len = PAGE_SIZE;
1201                 }
1202         }
1203
1204         if (cmd == OBD_BRW_WRITE) {
1205                 /* FIXME: need to clean up here */
1206                 rc = filter_journal_stop(journal_save, &obddev->u.filter,
1207                                          *desc_private);
1208         }
1209 out_ctxt:
1210         pop_ctxt(&saved);
1211         RETURN(rc);
1212 }
1213
1214 static int filter_write_locked_page(struct niobuf_local *lnb)
1215 {
1216         struct page *lpage;
1217         int rc;
1218
1219         lpage = lustre_get_page_write(lnb->dentry->d_inode, lnb->page->index);
1220         /* XXX */
1221
1222         memcpy(page_address(lpage), kmap(lnb->page), PAGE_SIZE);
1223         kunmap(lnb->page);
1224         __free_pages(lnb->page, 0);
1225
1226         rc = lustre_commit_page(lpage, 0, PAGE_SIZE);
1227         dput(lnb->dentry);
1228
1229         return rc;
1230 }
1231
1232 static int filter_commitrw(int cmd, struct obd_conn *conn,
1233                            int objcount, struct obd_ioobj *obj,
1234                            int niocount, struct niobuf_local *res,
1235                            void *private)
1236 {
1237         struct obd_run_ctxt saved;
1238         struct obd_ioobj *o = obj;
1239         struct niobuf_local *r = res;
1240         void *journal_save;
1241         int found_locked = 0;
1242         int i;
1243         ENTRY;
1244
1245         // if (cmd == OBD_BRW_WRITE)
1246         push_ctxt(&saved, &conn->oc_dev->u.filter.fo_ctxt);
1247         journal_save = current->journal_info;
1248         if (journal_save)
1249                 CERROR("Existing handle %p???\n", journal_save);
1250         current->journal_info = private;
1251         for (i = 0; i < objcount; i++, obj++) {
1252                 int j;
1253                 for (j = 0 ; j < o->ioo_bufcnt ; j++, r++) {
1254                         struct page *page = r->page;
1255
1256                         /* If there was an error setting up a particular page
1257                          * for I/O we still need to continue with the rest of
1258                          * the pages in order to balance prepate/commit_write
1259                          * calls, and to complete as much I/O as possible.
1260                          */
1261                         if (!page)
1262                                 LBUG();
1263
1264                         if (r->flags & N_LOCAL_TEMP_PAGE) {
1265                                 found_locked = 1;
1266                                 continue;
1267                         }
1268
1269                         if (cmd == OBD_BRW_WRITE) {
1270                                 int rc;
1271                                 rc = lustre_commit_page(page, 0, PAGE_SIZE);
1272
1273                                 /* FIXME: still need to iput the other inodes */
1274                                 if (rc)
1275                                         RETURN(rc);
1276                         } else
1277                                 lustre_put_page(page);
1278
1279                         CDEBUG(D_INODE,
1280                                "put inode %p (%ld), count = %d, nlink = %d\n",
1281                                page->mapping->host,
1282                                page->mapping->host->i_ino,
1283                                atomic_read(&page->mapping->host->i_count) - 1,
1284                                page->mapping->host->i_nlink);
1285                         dput(r->dentry);
1286                 }
1287         }
1288         if (!found_locked)
1289                 goto out;
1290
1291         for (i = 0; i < objcount; i++, obj++) {
1292                 int j;
1293                 for (j = 0 ; j < o->ioo_bufcnt ; j++, r++) {
1294                         int rc;
1295                         if (!(r->flags & N_LOCAL_TEMP_PAGE))
1296                                 continue;
1297
1298                         rc = filter_write_locked_page(r);
1299                         /* XXX */
1300                 }
1301         }
1302
1303 out:
1304         current->journal_info = journal_save;
1305         pop_ctxt(&saved);
1306         RETURN(0);
1307 }
1308
1309 static int filter_statfs(struct obd_conn *conn, struct statfs * statfs)
1310 {
1311         struct super_block *sb;
1312         int err;
1313         ENTRY;
1314
1315         if (!gen_client(conn)) {
1316                 CDEBUG(D_IOCTL, "invalid client %u\n", conn->oc_id);
1317                 RETURN(-EINVAL);
1318         }
1319
1320         sb = conn->oc_dev->u.filter.fo_sb;
1321
1322         err = sb->s_op->statfs(sb, statfs);
1323         RETURN(err);
1324 } /* filter_statfs */
1325
1326
1327 static int filter_get_info(struct obd_conn *conn, obd_count keylen,
1328                            void *key, obd_count *vallen, void **val)
1329 {
1330         struct obd_device *obddev;
1331         struct obd_client * cli;
1332         ENTRY;
1333
1334         if (!(cli = gen_client(conn))) {
1335                 CDEBUG(D_IOCTL, "invalid client %u\n", conn->oc_id);
1336                 RETURN(-EINVAL);
1337         }
1338
1339         obddev = conn->oc_dev;
1340
1341         if ( keylen == strlen("blocksize") &&
1342              memcmp(key, "blocksize", keylen) == 0 ) {
1343                 *vallen = sizeof(long);
1344                 *val = (void *)(long)obddev->u.filter.fo_sb->s_blocksize;
1345                 RETURN(0);
1346         }
1347
1348         if ( keylen == strlen("blocksize_bits") &&
1349              memcmp(key, "blocksize_bits", keylen) == 0 ){
1350                 *vallen = sizeof(long);
1351                 *val = (void *)(long)obddev->u.filter.fo_sb->s_blocksize_bits;
1352                 RETURN(0);
1353         }
1354
1355         if ( keylen == strlen("root_ino") &&
1356              memcmp(key, "root_ino", keylen) == 0 ){
1357                 *vallen = sizeof(long);
1358                 *val = (void *)(long)FILTER_ROOTINO;
1359                 RETURN(0);
1360         }
1361
1362         CDEBUG(D_IOCTL, "invalid key\n");
1363         RETURN(-EINVAL);
1364 }
1365
1366
1367 static struct obd_ops filter_obd_ops = {
1368         o_get_info:    filter_get_info,
1369         o_setup:       filter_setup,
1370         o_cleanup:     filter_cleanup,
1371         o_connect:     filter_connect,
1372         o_disconnect:  filter_disconnect,
1373         o_statfs:      filter_statfs,
1374         o_getattr:     filter_getattr,
1375         o_create:      filter_create,
1376         o_setattr:     filter_setattr,
1377         o_destroy:     filter_destroy,
1378         o_open:        filter_open,
1379         o_close:       filter_close,
1380         o_read:        filter_read,
1381         o_write:       filter_write,
1382         o_brw:         filter_pgcache_brw,
1383         o_punch:       filter_truncate,
1384         o_preprw:      filter_preprw,
1385         o_commitrw:    filter_commitrw
1386 #if 0
1387         o_preallocate: filter_preallocate_inodes,
1388         o_migrate:     filter_migrate,
1389         o_copy:        gen_copy_data,
1390         o_iterate:     filter_iterate
1391 #endif
1392 };
1393
1394
1395 static int __init obdfilter_init(void)
1396 {
1397         printk(KERN_INFO "Filtering OBD driver  v0.001, braam@clusterfs.com\n");
1398         return obd_register_type(&filter_obd_ops, OBD_FILTER_DEVICENAME);
1399 }
1400
1401 static void __exit obdfilter_exit(void)
1402 {
1403         obd_unregister_type(OBD_FILTER_DEVICENAME);
1404 }
1405
1406 MODULE_AUTHOR("Peter J. Braam <braam@clusterfs.com>");
1407 MODULE_DESCRIPTION("Lustre Filtering OBD driver v1.0");
1408 MODULE_LICENSE("GPL");
1409
1410 module_init(obdfilter_init);
1411 module_exit(obdfilter_exit);