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/ext2_obd/ext2_obd.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
  16 #include <linux/version.h>
  17 #include <linux/module.h>
  18 #include <linux/fs.h>
  19 #include <linux/stat.h>
  20 #include <linux/locks.h>
  21 #include <linux/ext2_fs.h>
  22 #include <linux/quotaops.h>
  23 #include <asm/unistd.h>
  24
  25 #define DEBUG_SUBSYSTEM S_FILTER
  26
  27 #include <linux/obd_class.h>
  28 #include <linux/obd_ext2.h>
  29 #include <linux/obd_filter.h>
  30
  31 extern struct obd_device obd_dev[MAX_OBD_DEVICES];
  32 long filter_memory;
  33
  34 #define FILTER_ROOTINO 2
  35
  36 #define S_SHIFT 12
  37 static char * obd_type_by_mode[S_IFMT >> S_SHIFT] = {
  38         [0]                     "",
  39         [S_IFREG >> S_SHIFT]    "R",
  40         [S_IFDIR >> S_SHIFT]    "D",
  41         [S_IFCHR >> S_SHIFT]    "C",
  42         [S_IFBLK >> S_SHIFT]    "B",
  43         [S_IFIFO >> S_SHIFT]    "F",
  44         [S_IFSOCK >> S_SHIFT]   "S",
  45         [S_IFLNK >> S_SHIFT]    "L"
  46 };
  47
  48
  49 /* write the pathname into the string */
  50 static void filter_id(char *buf, obd_id id, obd_mode mode)
  51 {
  52         sprintf(buf, "O/%s/%Ld", obd_type_by_mode[(mode & S_IFMT) >> S_SHIFT],
  53                 id);
  54 }
  55
  56 /* setup the object store with correct subdirectories */
  57 static int filter_prep(struct obd_device *obddev)
  58 {
  59         struct obd_run_ctxt saved;
  60         struct file *file;
  61         struct inode *inode;
  62         loff_t off;
  63         int rc = 0;
  64         char rootid[128];
  65         __u64 lastino = 2;
  66
  67         push_ctxt(&saved, &obddev->u.filter.fo_ctxt);
  68         rc = simple_mkdir(current->fs->pwd, "O", 0700);
  69         rc = simple_mkdir(current->fs->pwd, "P", 0700);
  70         rc = simple_mkdir(current->fs->pwd, "D", 0700);
  71         file = filp_open("O", O_RDONLY, 0);
  72         if (IS_ERR(file)) {
  73                 CERROR("cannot open O\n");
  74                 GOTO(out, rc = PTR_ERR(file));
  75         }
  76         rc = simple_mkdir(file->f_dentry, "R", 0700);  /* regular */
  77         rc = simple_mkdir(file->f_dentry, "D", 0700);  /* directory */
  78         rc = simple_mkdir(file->f_dentry, "L", 0700);  /* symbolic links */
  79         rc = simple_mkdir(file->f_dentry, "C", 0700);  /* character devices */
  80         rc = simple_mkdir(file->f_dentry, "B", 0700);  /* block devices */
  81         rc = simple_mkdir(file->f_dentry, "F", 0700);  /* fifo's */
  82         rc = simple_mkdir(file->f_dentry, "S", 0700);  /* sockets */
  83         filp_close(file, NULL);
  84
  85         filter_id(rootid, FILTER_ROOTINO, S_IFDIR);
  86         file = filp_open(rootid, O_RDWR | O_CREAT, 00755);
  87         if (IS_ERR(file)) {
  88                 CERROR("OBD filter: cannot make root directory");
  89                 GOTO(out, rc = PTR_ERR(file));
  90         }
  91         filp_close(file, 0);
  92         rc = simple_mkdir(current->fs->pwd, rootid, 0755);
  93
  94         file = filp_open("D/status", O_RDWR | O_CREAT, 0700);
  95         if ( !file || IS_ERR(file) ) {
  96                 CERROR("OBD filter: cannot open/create status file\n");
  97                 GOTO(out, rc = PTR_ERR(file));
  98         }
  99
 100         /* steal operations */
 101         inode = file->f_dentry->d_inode;
 102         obddev->u.filter.fo_fop = file->f_op;
 103         obddev->u.filter.fo_iop = inode->i_op;
 104         obddev->u.filter.fo_aops = inode->i_mapping->a_ops;
 105
 106         off = 0;
 107         if (inode->i_size == 0) {
 108                 ssize_t retval = file->f_op->write(file, (char *)&lastino,
 109                                                    sizeof(lastino), &off);
 110                 if (retval != sizeof(lastino)) {
 111                         CERROR("OBD filter: error writing lastino\n");
 112                         GOTO(out, rc = -EIO);
 113                 }
 114         } else {
 115                 ssize_t retval = file->f_op->read(file, (char *)&lastino,
 116                                                   sizeof(lastino), &off);
 117                 if (retval != sizeof(lastino)) {
 118                         CERROR("OBD filter: error reading lastino\n");
 119                         GOTO(out, rc = -EIO);
 120                 }
 121         }
 122         obddev->u.filter.fo_lastino = lastino;
 123         filp_close(file, 0);
 124
 125  out:
 126         pop_ctxt(&saved);
 127
 128         return(rc);
 129 }
 130
 131 /* cleanup the filter: write last used object id to status file */
 132 static void filter_post(struct obd_device *obddev)
 133 {
 134         struct obd_run_ctxt saved;
 135         long rc;
 136         struct file *file;
 137         loff_t off = 0;
 138
 139         push_ctxt(&saved, &obddev->u.filter.fo_ctxt);
 140         file = filp_open("D/status", O_RDWR | O_CREAT, 0700);
 141         if ( !file || IS_ERR(file)) {
 142                 CERROR("OBD filter: cannot create status file\n");
 143                 goto out;
 144         }
 145         rc = file->f_op->write(file, (char *)&obddev->u.filter.fo_lastino,
 146                        sizeof(obddev->u.filter.fo_lastino), &off);
 147         if (rc != sizeof(obddev->u.filter.fo_lastino) ) {
 148                 CERROR("OBD filter: error writing lastino\n");
 149         }
 150
 151         rc = filp_close(file, NULL);
 152         if (rc) {
 153                 CERROR("OBD filter: cannot close status file\n");
 154         }
 155  out:
 156         pop_ctxt(&saved);
 157 }
 158
 159
 160 static __u64 filter_next_id(struct obd_device *obddev)
 161 {
 162         __u64 id;
 163         spin_lock(&obddev->u.filter.fo_lock);
 164         obddev->u.filter.fo_lastino++;
 165         id =    obddev->u.filter.fo_lastino;
 166         spin_unlock(&obddev->u.filter.fo_lock);
 167         return id;
 168 }
 169
 170 /* how to get files, dentries, inodes from object id's */
 171 static struct file *filter_obj_open(struct obd_device *obddev,
 172                                    __u64 id, __u32 type)
 173 {
 174         struct obd_run_ctxt saved;
 175         char name[24];
 176         struct super_block *sb;
 177         struct file *file;
 178
 179         sb = obddev->u.filter.fo_sb;
 180         if (!sb || !sb->s_dev) {
 181                 CDEBUG(D_SUPER, "fatal: device not initialized.\n");
 182                 EXIT;
 183                 return NULL;
 184         }
 185
 186         if ( !id ) {
 187                 CDEBUG(D_INODE, "fatal: invalid obdo %Lu\n", id);
 188                 EXIT;
 189                 return NULL;
 190         }
 191
 192         if ( ! (type & S_IFMT) ) {
 193                 CERROR("OBD filter_obj_open, no type (%Ld), mode %o!\n",
 194                        id, type);
 195         }
 196
 197         filter_id(name, id, type);
 198         push_ctxt(&saved, &obddev->u.filter.fo_ctxt);
 199         file = filp_open(name, O_RDONLY | O_LARGEFILE, 0);
 200         pop_ctxt(&saved);
 201
 202         CDEBUG(D_INODE, "opening obdo %s\n", name);
 203
 204         return file;
 205 }
 206
 207 static struct file *filter_parent(obd_id id, obd_mode mode)
 208 {
 209         char path[64];
 210         struct file *file;
 211
 212         sprintf(path, "O/%s", obd_type_by_mode[(mode & S_IFMT) >> S_SHIFT]);
 213
 214         file = filp_open(path, O_RDONLY, 0);
 215         return file;
 216 }
 217
 218
 219 static struct inode *filter_inode_from_obj(struct obd_device *obddev,
 220                                      __u64 id, __u32 type)
 221 {
 222         struct file *file;
 223         struct inode *inode;
 224
 225         file = filter_obj_open(obddev, id, type);
 226         if ( !file ) {
 227                 CERROR("filter_inode_from_obdo failed\n");
 228                 return NULL;
 229         }
 230
 231         inode = iget(file->f_dentry->d_inode->i_sb,
 232                      file->f_dentry->d_inode->i_ino);
 233         filp_close(file, 0);
 234         return inode;
 235 }
 236
 237 /* obd methods */
 238 static int filter_connect(struct obd_conn *conn)
 239 {
 240         int rc;
 241
 242         MOD_INC_USE_COUNT;
 243         rc = gen_connect(conn);
 244
 245         if (rc)
 246                 MOD_DEC_USE_COUNT;
 247
 248         return rc;
 249 }
 250
 251 static int filter_disconnect(struct obd_conn *conn)
 252 {
 253         int rc;
 254
 255         rc = gen_disconnect(conn);
 256         if (!rc)
 257                 MOD_DEC_USE_COUNT;
 258
 259         /* XXX cleanup preallocated inodes */
 260         return rc;
 261 }
 262
 263 /* mount the file system (secretly) */
 264 static int filter_setup(struct obd_device *obddev, obd_count len, void *buf)
 265 {
 266         struct obd_ioctl_data* data = buf;
 267         struct vfsmount *mnt;
 268         int err = 0;
 269         ENTRY;
 270
 271         MOD_INC_USE_COUNT;
 272         mnt = do_kern_mount(data->ioc_inlbuf2, 0, data->ioc_inlbuf1, NULL);
 273         err = PTR_ERR(mnt);
 274         if (IS_ERR(mnt))
 275                 GOTO(err_dec, err);
 276
 277         /* XXX is this even possible if do_kern_mount succeeded? */
 278         obddev->u.filter.fo_sb = mnt->mnt_root->d_inode->i_sb;
 279         if (!obddev->u.filter.fo_sb)
 280                 GOTO(err_put, err = -ENODEV);
 281
 282         obddev->u.filter.fo_vfsmnt = mnt;
 283         obddev->u.filter.fo_fstype = strdup(data->ioc_inlbuf2);
 284
 285         obddev->u.filter.fo_ctxt.pwdmnt = mnt;
 286         obddev->u.filter.fo_ctxt.pwd = mnt->mnt_root;
 287         obddev->u.filter.fo_ctxt.fs = KERNEL_DS;
 288
 289         err = filter_prep(obddev);
 290         if (err)
 291                 GOTO(err_kfree, err);
 292         spin_lock_init(&obddev->u.filter.fo_lock);
 293
 294         RETURN(0);
 295
 296 err_kfree:
 297         kfree(obddev->u.filter.fo_fstype);
 298 err_put:
 299         unlock_kernel();
 300         mntput(obddev->u.filter.fo_vfsmnt);
 301         obddev->u.filter.fo_sb = 0;
 302         lock_kernel();
 303
 304 err_dec:
 305         MOD_DEC_USE_COUNT;
 306         return err;
 307 }
 308
 309
 310 static int filter_cleanup(struct obd_device * obddev)
 311 {
 312         struct super_block *sb;
 313
 314         ENTRY;
 315
 316         if ( !(obddev->obd_flags & OBD_SET_UP) )
 317                 RETURN(0);
 318
 319         if ( !list_empty(&obddev->obd_gen_clients) ) {
 320                 CERROR("still has clients!\n");
 321                 RETURN(-EBUSY);
 322         }
 323
 324         sb = obddev->u.filter.fo_sb;
 325         if (!obddev->u.filter.fo_sb)
 326                 RETURN(0);
 327
 328         filter_post(obddev);
 329
 330         unlock_kernel();
 331         mntput(obddev->u.filter.fo_vfsmnt);
 332         obddev->u.filter.fo_sb = 0;
 333         kfree(obddev->u.filter.fo_fstype);
 334
 335         lock_kernel();
 336
 337         MOD_DEC_USE_COUNT;
 338         RETURN(0);
 339 }
 340
 341
 342 static inline void filter_from_inode(struct obdo *oa, struct inode *inode)
 343 {
 344         int type = oa->o_mode & S_IFMT;
 345         ENTRY;
 346
 347         CDEBUG(D_INFO, "src inode %ld, dst obdo %ld valid 0x%08x\n",
 348                inode->i_ino, (long)oa->o_id, oa->o_valid);
 349         obdo_from_inode(oa, inode);
 350         oa->o_mode &= ~S_IFMT;
 351         oa->o_mode |= type;
 352
 353         if (S_ISCHR(inode->i_mode) || S_ISBLK(inode->i_mode)) {
 354                 obd_rdev rdev = kdev_t_to_nr(inode->i_rdev);
 355                 CDEBUG(D_INODE, "copying device %x from inode to obdo\n",
 356                        rdev);
 357                 *((obd_rdev *)oa->o_inline) = rdev;
 358                 oa->o_obdflags |= OBD_FL_INLINEDATA;
 359                 oa->o_valid |= OBD_MD_FLINLINE;
 360         }
 361
 362 #if 0
 363  else if (ext2obd_has_inline(inode)) {
 364                 CDEBUG(D_INFO, "copying inline from inode to obdo\n");
 365                 memcpy(oa->o_inline, inode->u.ext2_i.i_data,
 366                        MIN(sizeof(inode->u.ext2_i.i_data),OBD_INLINESZ));
 367                 oa->o_obdflags |= OBD_FL_INLINEDATA;
 368                 oa->o_valid |= OBD_MD_FLINLINE;
 369         }
 370
 371         if (ext2obd_has_obdmd(inode)) {
 372                 /* XXX this will change when we don't store the obdmd in data */
 373                 CDEBUG(D_INFO, "copying obdmd from inode to obdo\n");
 374                 memcpy(oa->o_obdmd, inode->u.ext2_i.i_data,
 375                        MIN(sizeof(inode->u.ext2_i.i_data),OBD_INLINESZ));
 376                 oa->o_obdflags |= OBD_FL_OBDMDEXISTS;
 377                 oa->o_valid |= OBD_MD_FLOBDMD;
 378         }
 379 #endif
 380         EXIT;
 381 }
 382
 383 static int filter_getattr(struct obd_conn *conn, struct obdo *oa)
 384 {
 385         struct inode *inode;
 386
 387         ENTRY;
 388         if ( !gen_client(conn) ) {
 389                 CDEBUG(D_IOCTL, "fatal: invalid client %u\n", conn->oc_id);
 390                 EXIT;
 391                 return -EINVAL;
 392         }
 393
 394         if ( !(inode = filter_inode_from_obj(conn->oc_dev,
 395                                               oa->o_id, oa->o_mode)) ) {
 396                 EXIT;
 397                 return -ENOENT;
 398         }
 399
 400         oa->o_valid &= ~OBD_MD_FLID;
 401         filter_from_inode(oa, inode);
 402
 403         iput(inode);
 404         EXIT;
 405         return 0;
 406 }
 407
 408 static int filter_setattr(struct obd_conn *conn, struct obdo *oa)
 409 {
 410         struct inode *inode;
 411         struct iattr iattr;
 412         int rc;
 413         struct dentry de;
 414
 415         if (!gen_client(conn)) {
 416                 CDEBUG(D_IOCTL, "invalid client %u\n", conn->oc_id);
 417                 return -EINVAL;
 418         }
 419
 420         inode = filter_inode_from_obj(conn->oc_dev, oa->o_id, oa->o_mode);
 421         if ( !inode ) {
 422                 EXIT;
 423                 return -ENOENT;
 424         }
 425
 426         iattr_from_obdo(&iattr, oa);
 427         iattr.ia_mode &= ~S_IFMT;
 428         iattr.ia_mode |= S_IFREG;
 429         de.d_inode = inode;
 430         if ( inode->i_op->setattr ) {
 431                 rc = inode->i_op->setattr(&de, &iattr);
 432         } else {
 433                 rc = inode_setattr(inode, &iattr);
 434         }
 435
 436         iput(inode);
 437         EXIT;
 438         return rc;
 439 }
 440
 441 static int filter_create (struct obd_conn* conn, struct obdo *oa)
 442 {
 443         char name[64];
 444         struct obd_run_ctxt saved;
 445         struct file *file;
 446         int mode;
 447         struct obd_device *obddev = conn->oc_dev;
 448         struct iattr;
 449         ENTRY;
 450
 451         if (!gen_client(conn)) {
 452                 CERROR("invalid client %u\n", conn->oc_id);
 453                 return -EINVAL;
 454         }
 455
 456         oa->o_id = filter_next_id(conn->oc_dev);
 457         if ( !(oa->o_mode && S_IFMT) ) {
 458                 CERROR("filter obd: no type!\n");
 459                 return -ENOENT;
 460         }
 461
 462         filter_id(name, oa->o_id, oa->o_mode);
 463         push_ctxt(&saved, &obddev->u.filter.fo_ctxt);
 464         mode = oa->o_mode;
 465         mode &= ~S_IFMT;
 466         mode |= S_IFREG;
 467         file = filp_open(name, O_RDONLY | O_CREAT, mode);
 468         pop_ctxt(&saved);
 469         if (IS_ERR(file)) {
 470                 CERROR("Error mknod obj %s, err %ld\n", name, PTR_ERR(file));
 471                 return -ENOENT;
 472         }
 473         filp_close(file, 0);
 474
 475         /* Set flags for fields we have set in ext2_new_inode */
 476         oa->o_valid |= OBD_MD_FLID | OBD_MD_FLBLKSZ | OBD_MD_FLBLOCKS |
 477                  OBD_MD_FLMTIME | OBD_MD_FLATIME | OBD_MD_FLCTIME |
 478                  OBD_MD_FLUID | OBD_MD_FLGID;
 479         return 0;
 480 }
 481
 482 static int filter_destroy(struct obd_conn *conn, struct obdo *oa)
 483 {
 484         struct obd_device * obddev;
 485         struct obd_client * cli;
 486         struct inode * inode;
 487         struct file *dir;
 488         struct file *object;
 489         int rc;
 490         struct obd_run_ctxt saved;
 491
 492         if (!(cli = gen_client(conn))) {
 493                 CERROR("invalid client %u\n", conn->oc_id);
 494                 EXIT;
 495                 return -EINVAL;
 496         }
 497
 498         obddev = conn->oc_dev;
 499         object = filter_obj_open(obddev, oa->o_id, oa->o_mode);
 500         if (!object || IS_ERR(object)) {
 501                 EXIT;
 502                 return -ENOENT;
 503         }
 504
 505         inode = object->f_dentry->d_inode;
 506         inode->i_nlink = 1;
 507         inode->i_mode = 010000;
 508
 509         push_ctxt(&saved, &obddev->u.filter.fo_ctxt);
 510         dir = filter_parent(oa->o_id, oa->o_mode);
 511         if (IS_ERR(dir)) {
 512                 rc = PTR_ERR(dir);
 513                 EXIT;
 514                 goto out;
 515         }
 516         dget(dir->f_dentry);
 517         dget(object->f_dentry);
 518         rc = vfs_unlink(dir->f_dentry->d_inode, object->f_dentry);
 519
 520         filp_close(dir, 0);
 521         filp_close(object, 0);
 522 out:
 523         pop_ctxt(&saved);
 524         EXIT;
 525         return rc;
 526 }
 527
 528 static int filter_truncate(struct obd_conn *conn, struct obdo *oa, obd_size count,
 529                          obd_off offset)
 530 {
 531         int error;
 532
 533         error = filter_setattr(conn, oa);
 534         oa->o_valid = OBD_MD_FLBLOCKS | OBD_MD_FLCTIME | OBD_MD_FLMTIME;
 535
 536         EXIT;
 537         return error;
 538 }
 539
 540 /* buffer must lie in user memory here */
 541 static int filter_read(struct obd_conn *conn, struct obdo *oa, char *buf,
 542                         obd_size *count, obd_off offset)
 543 {
 544         struct file * file;
 545         unsigned long retval;
 546         int err;
 547
 548         if (!gen_client(conn)) {
 549                 CDEBUG(D_IOCTL, "invalid client %u\n", conn->oc_id);
 550                 EXIT;
 551                 return -EINVAL;
 552         }
 553
 554         file = filter_obj_open(conn->oc_dev, oa->o_id, oa->o_mode);
 555         if (!file || IS_ERR(file)) {
 556                 EXIT;
 557                 return -PTR_ERR(file);
 558         }
 559
 560         /* count doubles as retval */
 561         retval = file->f_op->read(file, buf, *count, &offset);
 562         filp_close(file, 0);
 563
 564         if ( retval >= 0 ) {
 565                 err = 0;
 566                 *count = retval;
 567         } else {
 568                 err = retval;
 569                 *count = 0;
 570         }
 571
 572         return err;
 573 }
 574
 575
 576 /* buffer must lie in user memory here */
 577 static int filter_write(struct obd_conn *conn, struct obdo *oa, char *buf,
 578                          obd_size *count, obd_off offset)
 579 {
 580         int err;
 581         struct file * file;
 582         unsigned long retval;
 583
 584         ENTRY;
 585         if (!gen_client(conn)) {
 586                 CDEBUG(D_IOCTL, "invalid client %u\n", conn->oc_id);
 587                 EXIT;
 588                 return -EINVAL;
 589         }
 590
 591         file = filter_obj_open(conn->oc_dev, oa->o_id, oa->o_mode);
 592         if (!file || IS_ERR(file)) {
 593                 EXIT;
 594                 return -PTR_ERR(file);
 595         }
 596
 597         /* count doubles as retval */
 598         retval = file->f_op->write(file, buf, *count, &offset);
 599         filp_close(file, 0);
 600
 601         if ( retval >= 0 ) {
 602                 err = 0;
 603                 *count = retval;
 604                 EXIT;
 605         } else {
 606                 err = retval;
 607                 *count = 0;
 608                 EXIT;
 609         }
 610
 611         return err;
 612 } /* ext2obd_write */
 613
 614 static int filter_pgcache_brw(int rw, struct obd_conn *conn,
 615                                obd_count num_oa,
 616                                struct obdo **oa,
 617                                obd_count *oa_bufs,
 618                                struct page **pages,
 619                                obd_size *count,
 620                                obd_off *offset,
 621                                obd_flag *flags)
 622 {
 623         struct super_block      *sb;
 624         mm_segment_t oldfs;
 625         int                      onum;          /* index to oas */
 626         int                      pnum;          /* index to pages (bufs) */
 627         unsigned long            retval;
 628         int                      error;
 629         struct file *file;
 630
 631         ENTRY;
 632
 633         if (!gen_client(conn)) {
 634                 CDEBUG(D_IOCTL, "invalid client %u\n", conn->oc_id);
 635                 EXIT;
 636                 return -EINVAL;
 637         }
 638
 639         sb = conn->oc_dev->u.filter.fo_sb;
 640         oldfs = get_fs();
 641         set_fs(KERNEL_DS);
 642
 643         pnum = 0; /* pnum indexes buf 0..num_pages */
 644         for (onum = 0; onum < num_oa; onum++) {
 645                 int              pg;
 646
 647                 file = filter_obj_open(conn->oc_dev, oa[onum]->o_id,
 648                                        oa[onum]->o_mode);
 649                 if (!file || IS_ERR(file)) {
 650                         EXIT;
 651                         error = -ENOENT;
 652                         goto ERROR;
 653                 }
 654
 655                 /* count doubles as retval */
 656                 for (pg = 0; pg < oa_bufs[onum]; pg++) {
 657                         CDEBUG(D_INODE, "OP %d obdo no/pno: (%d,%d) (%ld,%ld) off count (%Ld,%Ld)\n",
 658                                rw, onum, pnum, file->f_dentry->d_inode->i_ino,
 659                                (unsigned long)offset[pnum] >> PAGE_CACHE_SHIFT,
 660                                offset[pnum], count[pnum]);
 661                         if (rw == WRITE) {
 662                                 loff_t off;
 663                                 char *buffer;
 664                                 off = offset[pnum];
 665                                 buffer = kmap(pages[pnum]);
 666                                 retval = file->f_op->write(file, buffer, count[pnum], &off);
 667                                 kunmap(pages[pnum]);
 668                                 CDEBUG(D_INODE, "retval %ld\n", retval);
 669                         } else {
 670                                 loff_t off = offset[pnum];
 671                                 char *buffer = kmap(pages[pnum]);
 672
 673                                 if (off >= file->f_dentry->d_inode->i_size) {
 674                                         memset(buffer, 0, count[pnum]);
 675                                         retval = count[pnum];
 676                                 } else {
 677                                         retval = file->f_op->read(file, buffer, count[pnum], &off);
 678                                 }
 679                                 kunmap(pages[pnum]);
 680
 681                                 if ( retval != count[pnum] ) {
 682                                         filp_close(file, 0);
 683                                         retval = -EIO;
 684                                         EXIT;
 685                                         goto ERROR;
 686                                 }
 687                                 CDEBUG(D_INODE, "retval %ld\n", retval);
 688                         }
 689                         pnum++;
 690                 }
 691                 /* sizes and blocks are set by generic_file_write */
 692                 /* ctimes/mtimes will follow with a setattr call */
 693                 filp_close(file, 0);
 694         }
 695
 696         EXIT;
 697  ERROR:
 698         set_fs(oldfs);
 699         error = (retval >= 0) ? 0 : retval;
 700         return error;
 701 }
 702
 703
 704 struct inode *ioobj_to_inode(struct obd_conn *conn, struct obd_ioobj *o)
 705 {
 706         struct inode *inode = NULL;
 707         struct super_block *sb = conn->oc_dev->u.ext2.e2_sb;
 708
 709         if (!sb || !sb->s_dev) {
 710                 CDEBUG(D_SUPER, "fatal: device not initialized.\n");
 711                 EXIT;
 712                 return NULL;
 713         }
 714
 715         if ( !o->ioo_id ) {
 716                 CDEBUG(D_INODE, "fatal: invalid obdo %lu\n", (long)o->ioo_id);
 717                 EXIT;
 718                 return NULL;
 719         }
 720
 721         inode = filter_inode_from_obj(conn->oc_dev, o->ioo_id, S_IFREG);
 722         if (!inode || inode->i_nlink == 0 || is_bad_inode(inode)) {
 723                 CERROR("from obdo - fatal: invalid inode %ld (%s).\n",
 724                        (long)o->ioo_id, inode ? inode->i_nlink ? "bad inode" :
 725                        "no links" : "NULL");
 726                 if (inode)
 727                         iput(inode);
 728                 EXIT;
 729                 return NULL;
 730         }
 731
 732         return inode;
 733 }
 734
 735 int filter_preprw(int cmd, struct obd_conn *conn,
 736                    int objcount, struct obd_ioobj *obj,
 737                    int niocount, struct niobuf *nb,
 738                    struct niobuf *res)
 739 {
 740         int i, j;
 741         struct obd_ioobj *o;
 742         struct niobuf *b;
 743         struct niobuf *r;
 744         struct inode *inode;
 745         struct page *page;
 746         int rc = 0;
 747         ENTRY;
 748
 749         memset(res, 0, sizeof(*res) * niocount);
 750
 751         for (i=0; i < objcount; i++) {
 752                 o = obj;
 753                 obj++;
 754                 for (j = 0 ; j < o->ioo_bufcnt ; j++) {
 755                         b = nb;
 756                         r = res;
 757                         nb++;
 758                         res++;
 759
 760                         inode = ioobj_to_inode(conn, o);
 761                         if (!inode) {
 762                                 EXIT;
 763                                 /* FIXME: we need to iput all of previous inodes */
 764                                 return -EINVAL;
 765                         }
 766
 767                         page = lustre_get_page(inode, b->offset >> PAGE_SHIFT);
 768                         if (IS_ERR(page)) {
 769                                 EXIT;
 770                                 return PTR_ERR(page);
 771                         }
 772                         if (cmd == OBD_BRW_WRITE) {
 773                                 rc = lustre_prepare_page(0, PAGE_SIZE, page);
 774                         }
 775                         if (rc) {
 776                                 CERROR("i %d j %d objcount %d bufcnt %d , rc %d, offset %Ld\n", i, j, objcount, o->ioo_bufcnt, rc, b->offset);
 777                         }
 778
 779                         r->addr = (__u64)(unsigned long)page_address(page);
 780                         r->offset = b->offset;
 781                         r->page = page;
 782                         r->len = PAGE_SIZE;
 783                         /* r->flags */
 784                 }
 785         }
 786         return 0;
 787 }
 788
 789 int filter_commitrw(int cmd, struct obd_conn *conn,
 790                      int objcount, struct obd_ioobj *obj,
 791                      int niocount, struct niobuf *res)
 792 {
 793         int i, j;
 794         int rc;
 795         struct inode *inode;
 796         struct obd_ioobj *o;
 797         struct niobuf *r = NULL;
 798         ENTRY;
 799
 800         for (i=0; i < objcount; i++) {
 801                 o = obj;
 802                 obj++;
 803
 804                 for (j = 0 ; j < o->ioo_bufcnt ; j++) {
 805                         r = res;
 806                         if (!r)
 807                                 LBUG();
 808                         res++;
 809
 810                         if (!r)
 811                                 LBUG();
 812
 813                         if (cmd == OBD_BRW_WRITE)
 814                                 rc = lustre_commit_page(r->page, 0, PAGE_SIZE);
 815                         else {
 816                                 lustre_put_page(r->page);
 817                                 rc = 0;
 818                         }
 819
 820                         if (rc) {
 821                                 EXIT;
 822                                 return rc;
 823                         }
 824                         inode = ((struct page *)r->page)->mapping->host;
 825                         iput(inode);
 826                 }
 827         }
 828         return 0;
 829 }
 830
 831 static int filter_statfs (struct obd_conn *conn, struct statfs * statfs)
 832 {
 833         struct super_block *sb;
 834         int err;
 835
 836         ENTRY;
 837
 838         if (!gen_client(conn)) {
 839                 CDEBUG(D_IOCTL, "invalid client %u\n", conn->oc_id);
 840                 EXIT;
 841                 return -EINVAL;
 842         }
 843
 844         sb = conn->oc_dev->u.filter.fo_sb;
 845
 846         err = sb->s_op->statfs(sb, statfs);
 847         EXIT;
 848         return err;
 849 } /* ext2obd_statfs */
 850
 851
 852 static int  filter_get_info(struct obd_conn *conn, obd_count keylen,
 853                              void *key, obd_count *vallen, void **val)
 854 {
 855         struct obd_device *obddev;
 856         struct obd_client * cli;
 857         ENTRY;
 858
 859         if (!(cli = gen_client(conn))) {
 860                 CDEBUG(D_IOCTL, "invalid client %u\n", conn->oc_id);
 861                 return -EINVAL;
 862         }
 863
 864         obddev = conn->oc_dev;
 865
 866         if ( keylen == strlen("blocksize") &&
 867              memcmp(key, "blocksize", keylen) == 0 ) {
 868                 *vallen = sizeof(int);
 869                 *val = (void *)obddev->u.filter.fo_sb->s_blocksize;
 870                 EXIT;
 871                 return 0;
 872         }
 873
 874         if ( keylen == strlen("blocksize_bits") &&
 875              memcmp(key, "blocksize_bits", keylen) == 0 ){
 876                 *vallen = sizeof(int);
 877                 *val = (void *)(int)obddev->u.filter.fo_sb->s_blocksize_bits;
 878                 EXIT;
 879                 return 0;
 880         }
 881
 882         if ( keylen == strlen("root_ino") &&
 883              memcmp(key, "root_ino", keylen) == 0 ){
 884                 *vallen = sizeof(int);
 885                 *val = (void *)(int) FILTER_ROOTINO;
 886                 EXIT;
 887                 return 0;
 888         }
 889
 890         CDEBUG(D_IOCTL, "invalid key\n");
 891         return -EINVAL;
 892 }
 893
 894
 895 struct obd_ops filter_obd_ops = {
 896         o_iocontrol:   NULL,
 897         o_get_info:    filter_get_info,
 898         o_setup:       filter_setup,
 899         o_cleanup:     filter_cleanup,
 900         o_connect:     filter_connect,
 901         o_disconnect:  filter_disconnect,
 902         o_statfs:      filter_statfs,
 903         o_getattr:     filter_getattr,
 904         o_create:      filter_create,
 905         o_setattr:     filter_setattr,
 906         o_destroy:     filter_destroy,
 907         o_read:        filter_read,
 908         o_write:       filter_write,
 909         o_brw:         filter_pgcache_brw,
 910         o_punch:       filter_truncate,
 911         o_preprw:      filter_preprw,
 912         o_commitrw:    filter_commitrw
 913 #if 0
 914         o_preallocate: ext2obd_preallocate_inodes,
 915         o_migrate:     ext2obd_migrate,
 916         o_copy:        gen_copy_data,
 917         o_iterate:     ext2obd_iterate
 918 #endif
 919 };
 920
 921
 922 static int __init obdfilter_init(void)
 923 {
 924         printk(KERN_INFO "Filtering OBD driver  v0.001, braam@clusterfs.com\n");
 925         return obd_register_type(&filter_obd_ops, OBD_FILTER_DEVICENAME);
 926 }
 927
 928 static void __exit obdfilter_exit(void)
 929 {
 930         obd_unregister_type(OBD_FILTER_DEVICENAME);
 931 }
 932
 933 MODULE_AUTHOR("Peter J. Braam <braam@clusterfs.com>");
 934 MODULE_DESCRIPTION("Lustre Filtering OBD driver v1.0");
 935 MODULE_LICENSE("GPL");
 936
 937 module_init(obdfilter_init);
 938 module_exit(obdfilter_exit);