lustre/llite/file.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (c) 2002, 2003 Cluster File Systems, Inc.
   5  *   Author: Peter Braam <braam@clusterfs.com>
   6  *   Author: Phil Schwan <phil@clusterfs.com>
   7  *   Author: Andreas Dilger <adilger@clusterfs.com>
   8  *
   9  *   This file is part of Lustre, http://www.lustre.org.
  10  *
  11  *   Lustre is free software; you can redistribute it and/or
  12  *   modify it under the terms of version 2 of the GNU General Public
  13  *   License as published by the Free Software Foundation.
  14  *
  15  *   Lustre is distributed in the hope that it will be useful,
  16  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  17  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  18  *   GNU General Public License for more details.
  19  *
  20  *   You should have received a copy of the GNU General Public License
  21  *   along with Lustre; if not, write to the Free Software
  22  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  23  */
  24
  25 #define DEBUG_SUBSYSTEM S_LLITE
  26 #include <linux/lustre_dlm.h>
  27 #include <linux/lustre_lite.h>
  28 #include <linux/pagemap.h>
  29 #include <linux/file.h>
  30 #include <linux/lustre_acl.h>
  31 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
  32 #include <linux/lustre_compat25.h>
  33 #endif
  34 #include "llite_internal.h"
  35 #include <linux/obd_lov.h>
  36
  37 int ll_md_och_close(struct obd_export *md_exp, struct inode *inode,
  38                     struct obd_client_handle *och)
  39 {
  40         struct ptlrpc_request *req = NULL;
  41         struct obdo *obdo = NULL;
  42         struct obd_device *obd;
  43         int rc;
  44         ENTRY;
  45
  46         obd = class_exp2obd(md_exp);
  47         if (obd == NULL) {
  48                 CERROR("Invalid MDC connection handle "LPX64"\n",
  49                        md_exp->exp_handle.h_cookie);
  50                 EXIT;
  51                 return 0;
  52         }
  53
  54         /*
  55          * here we check if this is forced umount. If so this is called on
  56          * canceling "open lock" and we do not call md_close() in this case , as
  57          * it will not successful, as import is already deactivated.
  58          */
  59         if (obd->obd_no_recov)
  60                 GOTO(out, rc = 0);
  61
  62         /* closing opened file */
  63         obdo = obdo_alloc();
  64         if (obdo == NULL)
  65                 RETURN(-ENOMEM);
  66
  67         obdo->o_id = inode->i_ino;
  68         obdo->o_valid = OBD_MD_FLID;
  69         obdo_from_inode(obdo, inode, (OBD_MD_FLTYPE | OBD_MD_FLMODE |
  70                                       OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
  71                                       OBD_MD_FLATIME | OBD_MD_FLMTIME |
  72                                       OBD_MD_FLCTIME));
  73         if (0 /* ll_is_inode_dirty(inode) */) {
  74                 obdo->o_flags = MDS_BFLAG_UNCOMMITTED_WRITES;
  75                 obdo->o_valid |= OBD_MD_FLFLAGS;
  76         }
  77         obdo->o_fid = id_fid(&ll_i2info(inode)->lli_id);
  78         obdo->o_mds = id_group(&ll_i2info(inode)->lli_id);
  79         rc = md_close(md_exp, obdo, och, &req);
  80         obdo_free(obdo);
  81
  82         if (rc == EAGAIN) {
  83                 /*
  84                  * we are the last writer, so the MDS has instructed us to get
  85                  * the file size and any write cookies, then close again.
  86                  */
  87
  88                 //ll_queue_done_writing(inode);
  89                 rc = 0;
  90         } else if (rc) {
  91                 CERROR("inode %lu mdc close failed: rc = %d\n",
  92                        (unsigned long)inode->i_ino, rc);
  93         }
  94
  95         ptlrpc_req_finished(req);
  96         EXIT;
  97 out:
  98         mdc_clear_open_replay_data(md_exp, och);
  99         och->och_fh.cookie = DEAD_HANDLE_MAGIC;
 100         OBD_FREE(och, sizeof *och);
 101         return rc;
 102 }
 103
 104 int ll_md_real_close(struct obd_export *md_exp,
 105                      struct inode *inode, int flags)
 106 {
 107         struct ll_inode_info *lli = ll_i2info(inode);
 108         struct obd_client_handle **och_p;
 109         struct obd_client_handle *och;
 110         __u64 *och_usecount;
 111         int rc = 0;
 112         ENTRY;
 113
 114         if (flags & FMODE_WRITE) {
 115                 och_p = &lli->lli_mds_write_och;
 116                 och_usecount = &lli->lli_open_fd_write_count;
 117         } else if (flags & FMODE_EXEC) {
 118                 och_p = &lli->lli_mds_exec_och;
 119                 och_usecount = &lli->lli_open_fd_exec_count;
 120          } else {
 121                 och_p = &lli->lli_mds_read_och;
 122                 och_usecount = &lli->lli_open_fd_read_count;
 123         }
 124
 125         down(&lli->lli_och_sem);
 126         if (*och_usecount) { /* There are still users of this handle, so
 127                                 skip freeing it. */
 128                 up(&lli->lli_och_sem);
 129                 RETURN(0);
 130         }
 131         och = *och_p;
 132
 133         *och_p = NULL;
 134         up(&lli->lli_och_sem);
 135
 136         /*
 137          * there might be a race and somebody have freed this och
 138          * already. Another way to have this twice called is if file closing
 139          * will fail due to netwok problems and on umount lock will be canceled
 140          * and this will be called from block_ast callack.
 141         */
 142         if (och && och->och_fh.cookie != DEAD_HANDLE_MAGIC)
 143                 rc = ll_md_och_close(md_exp, inode, och);
 144
 145         RETURN(rc);
 146 }
 147
 148 int ll_md_close(struct obd_export *md_exp, struct inode *inode,
 149                 struct file *file)
 150 {
 151         struct ll_file_data *fd = file->private_data;
 152         struct ll_inode_info *lli = ll_i2info(inode);
 153         int rc = 0;
 154         ENTRY;
 155
 156         /* clear group lock, if present */
 157         if (fd->fd_flags & LL_FILE_GROUP_LOCKED) {
 158                 struct lov_stripe_md *lsm = ll_i2info(inode)->lli_smd;
 159                 fd->fd_flags &= ~(LL_FILE_GROUP_LOCKED|LL_FILE_IGNORE_LOCK);
 160                 rc = ll_extent_unlock(fd, inode, lsm, LCK_GROUP,
 161                                       &fd->fd_cwlockh);
 162         }
 163
 164         /* Let's see if we have good enough OPEN lock on the file and if
 165            we can skip talking to MDS */
 166         if (file->f_dentry->d_inode) {
 167                 int lockmode;
 168                 struct obd_device *obddev;
 169                 struct lustre_handle lockh;
 170                 int flags = LDLM_FL_BLOCK_GRANTED;
 171                 struct ldlm_res_id file_res_id = {.name = {id_fid(&lli->lli_id),
 172                                                            id_group(&lli->lli_id)}};
 173                 ldlm_policy_data_t policy = {.l_inodebits={MDS_INODELOCK_OPEN}};
 174
 175                 down(&lli->lli_och_sem);
 176                 if (fd->fd_omode & FMODE_WRITE) {
 177                         lockmode = LCK_CW;
 178                         LASSERT(lli->lli_open_fd_write_count);
 179                         lli->lli_open_fd_write_count--;
 180                 } else if (fd->fd_omode & FMODE_EXEC) {
 181                         lockmode = LCK_PR;
 182                         LASSERT(lli->lli_open_fd_exec_count);
 183                         lli->lli_open_fd_exec_count--;
 184                 } else {
 185                         lockmode = LCK_CR;
 186                         LASSERT(lli->lli_open_fd_read_count);
 187                         lli->lli_open_fd_read_count--;
 188                 }
 189                 up(&lli->lli_och_sem);
 190
 191                 obddev = md_get_real_obd(md_exp, &lli->lli_id);
 192                 if (!ldlm_lock_match(obddev->obd_namespace, flags, &file_res_id,
 193                                      LDLM_IBITS, &policy, lockmode, &lockh))
 194                 {
 195                         rc = ll_md_real_close(md_exp, file->f_dentry->d_inode,
 196                                               fd->fd_omode);
 197                 } else {
 198                         ldlm_lock_decref(&lockh, lockmode);
 199                 }
 200         }
 201
 202         file->private_data = NULL;
 203         OBD_SLAB_FREE(fd, ll_file_data_slab, sizeof(*fd));
 204         RETURN(rc);
 205 }
 206
 207 /* While this returns an error code, fput() the caller does not, so we need
 208  * to make every effort to clean up all of our state here.  Also, applications
 209  * rarely check close errors and even if an error is returned they will not
 210  * re-try the close call.
 211  */
 212 int ll_file_release(struct inode *inode, struct file *file)
 213 {
 214         struct ll_file_data *fd;
 215         struct ll_sb_info *sbi = ll_i2sbi(inode);
 216         int rc;
 217
 218         ENTRY;
 219         CDEBUG(D_VFSTRACE, "VFS Op:inode="DLID4"(%p)\n",
 220                OLID4(&ll_i2info(inode)->lli_id), inode);
 221
 222         /* don't do anything for / */
 223         if (inode->i_sb->s_root == file->f_dentry)
 224                 RETURN(0);
 225
 226         lprocfs_counter_incr(sbi->ll_stats, LPROC_LL_RELEASE);
 227         fd = (struct ll_file_data *)file->private_data;
 228         LASSERT(fd != NULL);
 229
 230         rc = ll_md_close(sbi->ll_md_exp, inode, file);
 231         RETURN(rc);
 232 }
 233
 234 static int ll_intent_file_open(struct file *file, void *lmm,
 235                                int lmmsize, struct lookup_intent *itp)
 236 {
 237         struct ll_sb_info *sbi = ll_i2sbi(file->f_dentry->d_inode);
 238         const char *name = (char *)file->f_dentry->d_name.name;
 239         struct dentry *parent = file->f_dentry->d_parent;
 240         const int len = file->f_dentry->d_name.len;
 241         struct lustre_handle lockh;
 242         struct mdc_op_data *op_data;
 243         int rc;
 244
 245         if (!parent)
 246                 RETURN(-ENOENT);
 247
 248         OBD_ALLOC(op_data, sizeof(*op_data));
 249         if (op_data == NULL)
 250                 RETURN(-ENOMEM);
 251
 252         ll_prepare_mdc_data(op_data, parent->d_inode, NULL,
 253                             name, len, O_RDWR);
 254
 255         rc = md_enqueue(sbi->ll_md_exp, LDLM_IBITS, itp, LCK_PR, op_data,
 256                         &lockh, lmm, lmmsize, ldlm_completion_ast,
 257                         ll_mdc_blocking_ast, NULL);
 258         OBD_FREE(op_data, sizeof(*op_data));
 259         if (rc == 0) {
 260                 if (LUSTRE_IT(itp)->it_lock_mode)
 261                         memcpy(&LUSTRE_IT(itp)->it_lock_handle,
 262                                &lockh, sizeof(lockh));
 263
 264         } else if (rc < 0) {
 265                 CERROR("lock enqueue: err: %d\n", rc);
 266         }
 267         RETURN(rc);
 268 }
 269
 270 void ll_och_fill(struct inode *inode, struct lookup_intent *it,
 271                  struct obd_client_handle *och)
 272 {
 273         struct ptlrpc_request *req = LUSTRE_IT(it)->it_data;
 274         struct ll_inode_info *lli = ll_i2info(inode);
 275         struct mds_body *body;
 276         LASSERT(och);
 277
 278         body = lustre_msg_buf (req->rq_repmsg, 1, sizeof (*body));
 279         LASSERT (body != NULL);          /* reply already checked out */
 280         LASSERT_REPSWABBED (req, 1);     /* and swabbed down */
 281
 282         memcpy(&och->och_fh, &body->handle, sizeof(body->handle));
 283         och->och_magic = OBD_CLIENT_HANDLE_MAGIC;
 284         lli->lli_io_epoch = body->io_epoch;
 285         mdc_set_open_replay_data(ll_i2mdexp(inode), och,
 286                                  LUSTRE_IT(it)->it_data);
 287 }
 288
 289 int ll_local_open(struct file *file, struct lookup_intent *it,
 290                   struct obd_client_handle *och)
 291 {
 292         struct ll_file_data *fd;
 293         ENTRY;
 294
 295         if (och)
 296                 ll_och_fill(file->f_dentry->d_inode, it, och);
 297
 298         LASSERTF(file->private_data == NULL, "file %.*s/%.*s ino %lu/%u (%o)\n",
 299                  file->f_dentry->d_name.len, file->f_dentry->d_name.name,
 300                  file->f_dentry->d_parent->d_name.len,
 301                  file->f_dentry->d_parent->d_name.name,
 302                  file->f_dentry->d_inode->i_ino,
 303                  file->f_dentry->d_inode->i_generation,
 304                  file->f_dentry->d_inode->i_mode);
 305
 306         OBD_SLAB_ALLOC(fd, ll_file_data_slab, SLAB_KERNEL, sizeof *fd);
 307
 308         /* We can't handle this well without reorganizing ll_file_open and
 309          * ll_md_close(), so don't even try right now. */
 310         LASSERT(fd != NULL);
 311
 312         file->private_data = fd;
 313         ll_readahead_init(file->f_dentry->d_inode, &fd->fd_ras);
 314         fd->fd_omode = it->it_flags;
 315         RETURN(0);
 316 }
 317
 318 /* Open a file, and (for the very first open) create objects on the OSTs at
 319  * this time.  If opened with O_LOV_DELAY_CREATE, then we don't do the object
 320  * creation or open until ll_lov_setstripe() ioctl is called.  We grab
 321  * lli_open_sem to ensure no other process will create objects, send the
 322  * stripe MD to the MDS, or try to destroy the objects if that fails.
 323  *
 324  * If we already have the stripe MD locally then we don't request it in
 325  * mdc_open(), by passing a lmm_size = 0.
 326  *
 327  * It is up to the application to ensure no other processes open this file
 328  * in the O_LOV_DELAY_CREATE case, or the default striping pattern will be
 329  * used.  We might be able to avoid races of that sort by getting lli_open_sem
 330  * before returning in the O_LOV_DELAY_CREATE case and dropping it here
 331  * or in ll_file_release(), but I'm not sure that is desirable/necessary.
 332  */
 333 int ll_file_open(struct inode *inode, struct file *file)
 334 {
 335         struct ll_inode_info *lli = ll_i2info(inode);
 336         struct lookup_intent *it, oit = { .it_op = IT_OPEN,
 337                                           .it_flags = file->f_flags };
 338         struct lov_stripe_md *lsm;
 339         struct ptlrpc_request *req;
 340         int rc = 0;
 341         struct obd_client_handle **och_p;
 342         __u64 *och_usecount;
 343         ENTRY;
 344
 345         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p), flags %o\n",
 346                inode->i_ino, inode->i_generation, inode, file->f_flags);
 347
 348         /* don't do anything for / */
 349         if (inode->i_sb->s_root == file->f_dentry)
 350                 RETURN(0);
 351
 352         if ((file->f_flags+1) & O_ACCMODE)
 353                 oit.it_flags++;
 354         if (file->f_flags & O_TRUNC)
 355                 oit.it_flags |= 2;
 356
 357         it = file->f_it;
 358
 359         /*
 360          * sometimes LUSTRE_IT(it) may not be allocated like opening file by
 361          * dentry_open() from GNS stuff.
 362          */
 363         if (!it || !LUSTRE_IT(it)) {
 364                 it = &oit;
 365                 rc = ll_intent_alloc(it);
 366                 if (rc)
 367                         GOTO(out, rc);
 368         }
 369
 370         lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats, LPROC_LL_OPEN);
 371
 372         /*
 373          * mdc_intent_lock() didn't get a request ref if there was an open
 374          * error, so don't do cleanup on the * request here (bug 3430)
 375          */
 376         if (LUSTRE_IT(it)->it_disposition) {
 377                 rc = it_open_error(DISP_OPEN_OPEN, it);
 378                 if (rc)
 379                         RETURN(rc);
 380         }
 381
 382         /* Let's see if we have file open on MDS already. */
 383         if (it->it_flags & FMODE_WRITE) {
 384                 och_p = &lli->lli_mds_write_och;
 385                 och_usecount = &lli->lli_open_fd_write_count;
 386         } else if (it->it_flags & FMODE_EXEC) {
 387                 och_p = &lli->lli_mds_exec_och;
 388                 och_usecount = &lli->lli_open_fd_exec_count;
 389         } else {
 390                 och_p = &lli->lli_mds_read_och;
 391                 och_usecount = &lli->lli_open_fd_read_count;
 392         }
 393
 394         down(&lli->lli_och_sem);
 395         if (*och_p) { /* Open handle is present */
 396                 if (LUSTRE_IT(it)->it_disposition) {
 397                         struct obd_client_handle *och;
 398                         /* Well, there's extra open request that we do not need,
 399                            let's close it somehow*/
 400                         OBD_ALLOC(och, sizeof (struct obd_client_handle));
 401                         if (!och) {
 402                                 up(&lli->lli_och_sem);
 403                                 RETURN(-ENOMEM);
 404                         }
 405
 406                         ll_och_fill(inode, it, och);
 407                         /* ll_md_och_close() will free och */
 408                         ll_md_och_close(ll_i2mdexp(inode), inode, och);
 409                 }
 410                 (*och_usecount)++;
 411
 412                 rc = ll_local_open(file, it, NULL);
 413                 if (rc)
 414                         LBUG();
 415         } else {
 416                 LASSERT(*och_usecount == 0);
 417                 OBD_ALLOC(*och_p, sizeof (struct obd_client_handle));
 418                 if (!*och_p)
 419                         GOTO(out, rc = -ENOMEM);
 420                 (*och_usecount)++;
 421
 422                 if (!it || !LUSTRE_IT(it) || !LUSTRE_IT(it)->it_disposition) {
 423                         /*
 424                          * we are going to replace intent here, and that may
 425                          * possibly change access mode (FMODE_EXEC can only be
 426                          * set in intent), but I hope it never happens (I was
 427                          * not able to trigger it yet at least) -- green
 428                          */
 429
 430                         /* FIXME: FMODE_EXEC is not covered by O_ACCMODE! */
 431                         LASSERT(!(it->it_flags & FMODE_EXEC));
 432                         LASSERTF((it->it_flags & O_ACCMODE) ==
 433                                  (oit.it_flags & O_ACCMODE), "Changing intent "
 434                                  "flags %x to incompatible %x\n", it->it_flags,
 435                                  oit.it_flags);
 436                         it = &oit;
 437                         rc = ll_intent_file_open(file, NULL, 0, it);
 438                         if (rc)
 439                                 GOTO(out, rc);
 440                         rc = it_open_error(DISP_OPEN_OPEN, it);
 441                         if (rc)
 442                                 GOTO(out_och_free, rc);
 443
 444                         mdc_set_lock_data(NULL, &LUSTRE_IT(it)->it_lock_handle,
 445                                           file->f_dentry->d_inode);
 446                 }
 447                 lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats, LPROC_LL_OPEN);
 448                 rc = ll_local_open(file, it, *och_p);
 449                 LASSERTF(rc == 0, "rc = %d\n", rc);
 450         }
 451         up(&lli->lli_och_sem);
 452
 453         /*
 454          * must do this outside lli_och_sem lock to prevent deadlock where
 455          * different kind of OPEN lock for this same inode gets cancelled by
 456          * ldlm_cancel_lru
 457          */
 458
 459         if (!S_ISREG(inode->i_mode))
 460                 GOTO(out, rc);
 461
 462         lsm = lli->lli_smd;
 463         if (lsm == NULL) {
 464                 if (file->f_flags & O_LOV_DELAY_CREATE ||
 465                     !(file->f_mode & FMODE_WRITE)) {
 466                         CDEBUG(D_INODE, "object creation was delayed\n");
 467                         GOTO(out, rc);
 468                 }
 469         }
 470         file->f_flags &= ~O_LOV_DELAY_CREATE;
 471         GOTO(out, rc);
 472  out:
 473         req = LUSTRE_IT(it)->it_data;
 474         ll_intent_drop_lock(it);
 475         ll_intent_release(it);
 476         ptlrpc_req_finished(req);
 477         if (rc == 0) {
 478                 ll_open_complete(inode);
 479         } else {
 480 out_och_free:
 481                 if (*och_p) {
 482                         OBD_FREE(*och_p, sizeof (struct obd_client_handle));
 483                         *och_p = NULL; /* OBD_FREE writes some magic there */
 484                         (*och_usecount)--;
 485                 }
 486                 up(&lli->lli_och_sem);
 487         }
 488
 489         return rc;
 490 }
 491
 492 /* Fills the obdo with the attributes for the inode defined by lsm */
 493 int ll_lsm_getattr(struct obd_export *exp, struct lov_stripe_md *lsm,
 494                    struct obdo *oa)
 495 {
 496         struct ptlrpc_request_set *set;
 497         int rc;
 498         ENTRY;
 499
 500         LASSERT(lsm != NULL);
 501
 502         memset(oa, 0, sizeof *oa);
 503         oa->o_id = lsm->lsm_object_id;
 504         oa->o_gr = lsm->lsm_object_gr;
 505         oa->o_mode = S_IFREG;
 506         oa->o_valid = OBD_MD_FLID | OBD_MD_FLTYPE | OBD_MD_FLSIZE |
 507                 OBD_MD_FLBLOCKS | OBD_MD_FLBLKSZ | OBD_MD_FLMTIME |
 508                 OBD_MD_FLCTIME | OBD_MD_FLGROUP;
 509
 510         set = ptlrpc_prep_set();
 511         if (set == NULL) {
 512                 rc = -ENOMEM;
 513         } else {
 514                 rc = obd_getattr_async(exp, oa, lsm, set);
 515                 if (rc == 0)
 516                         rc = ptlrpc_set_wait(set);
 517                 ptlrpc_set_destroy(set);
 518         }
 519         if (rc)
 520                 RETURN(rc);
 521
 522         oa->o_valid &= (OBD_MD_FLBLOCKS | OBD_MD_FLBLKSZ | OBD_MD_FLMTIME |
 523                         OBD_MD_FLCTIME | OBD_MD_FLSIZE);
 524         RETURN(0);
 525 }
 526
 527 static inline void ll_remove_suid(struct inode *inode)
 528 {
 529         unsigned int mode;
 530
 531         /* set S_IGID if S_IXGRP is set, and always set S_ISUID */
 532         mode = (inode->i_mode & S_IXGRP)*(S_ISGID/S_IXGRP) | S_ISUID;
 533
 534         /* was any of the uid bits set? */
 535         mode &= inode->i_mode;
 536         if (mode && !capable(CAP_FSETID)) {
 537                 inode->i_mode &= ~mode;
 538                 // XXX careful here - we cannot change the size
 539         }
 540 }
 541
 542 static int ll_lock_to_stripe_offset(struct inode *inode, struct ldlm_lock *lock)
 543 {
 544         struct ll_inode_info *lli = ll_i2info(inode);
 545         struct lov_stripe_md *lsm = lli->lli_smd;
 546         struct obd_export *exp = ll_i2dtexp(inode);
 547         struct {
 548                 char name[16];
 549                 struct ldlm_lock *lock;
 550                 struct lov_stripe_md *lsm;
 551         } key = { .name = "lock_to_stripe", .lock = lock, .lsm = lsm };
 552         __u32 stripe, vallen = sizeof(stripe);
 553         int rc;
 554         ENTRY;
 555
 556         if (lsm->lsm_stripe_count == 1)
 557                 GOTO(check, stripe = 0);
 558
 559         /* get our offset in the lov */
 560         rc = obd_get_info(exp, sizeof(key), &key, &vallen, &stripe);
 561         if (rc != 0) {
 562                 CERROR("obd_get_info: rc = %d\n", rc);
 563                 RETURN(rc);
 564         }
 565         LASSERT(stripe < lsm->lsm_stripe_count);
 566         EXIT;
 567 check:
 568         if (lsm->lsm_oinfo[stripe].loi_id != lock->l_resource->lr_name.name[0]||
 569             lsm->lsm_oinfo[stripe].loi_gr != lock->l_resource->lr_name.name[2]){
 570                 LDLM_ERROR(lock, "resource doesn't match object "LPU64"/"LPU64
 571                            " inode=%lu/%u (%p)\n",
 572                            lsm->lsm_oinfo[stripe].loi_id,
 573                            lsm->lsm_oinfo[stripe].loi_gr,
 574                            inode->i_ino, inode->i_generation, inode);
 575                 return -ELDLM_NO_LOCK_DATA;
 576         }
 577
 578         return stripe;
 579 }
 580
 581 /* Flush the page cache for an extent as its canceled.  When we're on an LOV,
 582  * we get a lock cancellation for each stripe, so we have to map the obd's
 583  * region back onto the stripes in the file that it held.
 584  *
 585  * No one can dirty the extent until we've finished our work and they can
 586  * enqueue another lock.  The DLM protects us from ll_file_read/write here,
 587  * but other kernel actors could have pages locked.
 588  *
 589  * Called with the DLM lock held. */
 590 void ll_pgcache_remove_extent(struct inode *inode, struct lov_stripe_md *lsm,
 591                               struct ldlm_lock *lock, __u32 stripe)
 592 {
 593         ldlm_policy_data_t tmpex;
 594         unsigned long start, end, count, skip, i, j;
 595         struct page *page;
 596         int rc, rc2, discard = lock->l_flags & LDLM_FL_DISCARD_DATA;
 597         struct lustre_handle lockh;
 598         ENTRY;
 599
 600         memcpy(&tmpex, &lock->l_policy_data, sizeof(tmpex));
 601         CDEBUG(D_INODE|D_PAGE, "inode %lu(%p) ["LPU64"->"LPU64"] size: %llu\n",
 602                inode->i_ino, inode, tmpex.l_extent.start, tmpex.l_extent.end,
 603                inode->i_size);
 604
 605         /* our locks are page granular thanks to osc_enqueue, we invalidate the
 606          * whole page. */
 607         LASSERT((tmpex.l_extent.start & ~PAGE_CACHE_MASK) == 0);
 608         LASSERT(((tmpex.l_extent.end + 1) & ~PAGE_CACHE_MASK) == 0);
 609
 610         count = ~0;
 611         skip = 0;
 612         start = tmpex.l_extent.start >> PAGE_CACHE_SHIFT;
 613         end = tmpex.l_extent.end >> PAGE_CACHE_SHIFT;
 614         if (lsm->lsm_stripe_count > 1) {
 615                 count = lsm->lsm_stripe_size >> PAGE_CACHE_SHIFT;
 616                 skip = (lsm->lsm_stripe_count - 1) * count;
 617                 start += start/count * skip + stripe * count;
 618                 if (end != ~0)
 619                         end += end/count * skip + stripe * count;
 620         }
 621         if (end < tmpex.l_extent.end >> PAGE_CACHE_SHIFT)
 622                 end = ~0;
 623
 624         i = inode->i_size ? (inode->i_size - 1) >> PAGE_CACHE_SHIFT : 0;
 625         if (i < end)
 626                 end = i;
 627
 628         CDEBUG(D_INODE|D_PAGE, "walking page indices start: %lu j: %lu "
 629                "count: %lu skip: %lu end: %lu%s\n", start, start % count,
 630                count, skip, end, discard ? " (DISCARDING)" : "");
 631
 632         /* walk through the vmas on the inode and tear down mmaped pages that
 633          * intersect with the lock.  this stops immediately if there are no
 634          * mmap()ed regions of the file.  This is not efficient at all and
 635          * should be short lived. We'll associate mmap()ed pages with the lock
 636          * and will be able to find them directly */
 637
 638         for (i = start; i <= end; i += (j + skip)) {
 639                 j = min(count - (i % count), end - i + 1);
 640                 LASSERT(j > 0);
 641                 LASSERT(inode->i_mapping);
 642                 if (ll_teardown_mmaps(inode->i_mapping, i << PAGE_CACHE_SHIFT,
 643                                       ((i+j) << PAGE_CACHE_SHIFT) - 1) )
 644                         break;
 645         }
 646
 647         /* this is the simplistic implementation of page eviction at
 648          * cancelation.  It is careful to get races with other page
 649          * lockers handled correctly.  fixes from bug 20 will make it
 650          * more efficient by associating locks with pages and with
 651          * batching writeback under the lock explicitly. */
 652         for (i = start, j = start % count; i <= end;
 653              j++, i++, tmpex.l_extent.start += PAGE_CACHE_SIZE) {
 654                 if (j == count) {
 655                         CDEBUG(D_PAGE, "skip index %lu to %lu\n", i, i + skip);
 656                         i += skip;
 657                         j = 0;
 658                         if (i > end)
 659                                 break;
 660                 }
 661                 LASSERTF(tmpex.l_extent.start< lock->l_policy_data.l_extent.end,
 662                          LPU64" >= "LPU64" start %lu i %lu end %lu\n",
 663                          tmpex.l_extent.start, lock->l_policy_data.l_extent.end,
 664                          start, i, end);
 665
 666                 if (!mapping_has_pages(inode->i_mapping)) {
 667                         CDEBUG(D_INODE|D_PAGE, "nothing left\n");
 668                         break;
 669                 }
 670
 671                 cond_resched();
 672
 673                 page = find_get_page(inode->i_mapping, i);
 674                 if (page == NULL)
 675                         continue;
 676                 LL_CDEBUG_PAGE(D_PAGE, page, "lock page idx %lu ext "LPU64"\n",
 677                                i, tmpex.l_extent.start);
 678                 lock_page(page);
 679
 680                 /* page->mapping to check with racing against teardown */
 681                 if (!discard && clear_page_dirty_for_io(page)) {
 682                         rc = ll_call_writepage(inode, page);
 683                         if (rc != 0)
 684                                 CERROR("writepage of page %p failed: %d\n",
 685                                        page, rc);
 686                         /* either waiting for io to complete or reacquiring
 687                          * the lock that the failed writepage released */
 688                         lock_page(page);
 689                 }
 690
 691                 tmpex.l_extent.end = tmpex.l_extent.start + PAGE_CACHE_SIZE - 1;
 692                 /* check to see if another DLM lock covers this page */
 693                 rc2 = ldlm_lock_match(lock->l_resource->lr_namespace,
 694                                       LDLM_FL_BLOCK_GRANTED|LDLM_FL_CBPENDING |
 695                                       LDLM_FL_TEST_LOCK,
 696                                       &lock->l_resource->lr_name, LDLM_EXTENT,
 697                                       &tmpex, LCK_PR | LCK_PW, &lockh);
 698                 if (rc2 == 0 && page->mapping != NULL) {
 699                         // checking again to account for writeback's lock_page()
 700                         LL_CDEBUG_PAGE(D_PAGE, page, "truncating\n");
 701                         ll_ra_accounting(page, inode->i_mapping);
 702                         ll_truncate_complete_page(page);
 703                 }
 704                 unlock_page(page);
 705                 page_cache_release(page);
 706         }
 707         LASSERTF(tmpex.l_extent.start <=
 708                  (lock->l_policy_data.l_extent.end == ~0ULL ? ~0ULL :
 709                   lock->l_policy_data.l_extent.end + 1),
 710                  "loop too long "LPU64" > "LPU64" start %lu i %lu end %lu\n",
 711                  tmpex.l_extent.start, lock->l_policy_data.l_extent.end,
 712                  start, i, end);
 713         EXIT;
 714 }
 715
 716 static int ll_extent_lock_callback(struct ldlm_lock *lock,
 717                                    struct ldlm_lock_desc *new, void *data,
 718                                    int flag)
 719 {
 720         struct lustre_handle lockh = { 0 };
 721         int rc;
 722         ENTRY;
 723
 724         if ((unsigned long)data > 0 && (unsigned long)data < 0x1000) {
 725                 LDLM_ERROR(lock, "cancelling lock with bad data %p", data);
 726                 LBUG();
 727         }
 728
 729         switch (flag) {
 730         case LDLM_CB_BLOCKING:
 731                 ldlm_lock2handle(lock, &lockh);
 732                 rc = ldlm_cli_cancel(&lockh);
 733                 if (rc != ELDLM_OK)
 734                         CERROR("ldlm_cli_cancel failed: %d\n", rc);
 735                 break;
 736         case LDLM_CB_CANCELING: {
 737                 struct inode *inode;
 738                 struct ll_inode_info *lli;
 739                 struct lov_stripe_md *lsm;
 740                 __u32 stripe;
 741                 __u64 kms;
 742
 743                 /* This lock wasn't granted, don't try to evict pages */
 744                 if (lock->l_req_mode != lock->l_granted_mode)
 745                         RETURN(0);
 746
 747                 inode = ll_inode_from_lock(lock);
 748                 if (inode == NULL)
 749                         RETURN(0);
 750                 lli = ll_i2info(inode);
 751                 if (lli == NULL)
 752                         goto iput;
 753                 if (lli->lli_smd == NULL)
 754                         goto iput;
 755                 lsm = lli->lli_smd;
 756
 757                 stripe = ll_lock_to_stripe_offset(inode, lock);
 758                 if (stripe < 0)
 759                         goto iput;
 760                 ll_pgcache_remove_extent(inode, lsm, lock, stripe);
 761
 762                 l_lock(&lock->l_resource->lr_namespace->ns_lock);
 763                 down(&lli->lli_size_sem);
 764                 kms = ldlm_extent_shift_kms(lock,
 765                                             lsm->lsm_oinfo[stripe].loi_kms);
 766
 767                 if (lsm->lsm_oinfo[stripe].loi_kms != kms)
 768                         LDLM_DEBUG(lock, "updating kms from "LPU64" to "LPU64,
 769                                    lsm->lsm_oinfo[stripe].loi_kms, kms);
 770                 lsm->lsm_oinfo[stripe].loi_kms = kms;
 771                 up(&lli->lli_size_sem);
 772                 l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 773                 //ll_try_done_writing(inode);
 774         iput:
 775                 iput(inode);
 776                 break;
 777         }
 778         default:
 779                 LBUG();
 780         }
 781
 782         RETURN(0);
 783 }
 784
 785 #if 0
 786 int ll_async_completion_ast(struct ldlm_lock *lock, int flags, void *data)
 787 {
 788         /* XXX ALLOCATE - 160 bytes */
 789         struct inode *inode = ll_inode_from_lock(lock);
 790         struct ll_inode_info *lli = ll_i2info(inode);
 791         struct lustre_handle lockh = { 0 };
 792         struct ost_lvb *lvb;
 793         __u32 stripe;
 794         ENTRY;
 795
 796         if (flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED |
 797                      LDLM_FL_BLOCK_CONV)) {
 798                 LBUG(); /* not expecting any blocked async locks yet */
 799                 LDLM_DEBUG(lock, "client-side async enqueue returned a blocked "
 800                            "lock, returning");
 801                 ldlm_lock_dump(D_OTHER, lock, 0);
 802                 ldlm_reprocess_all(lock->l_resource);
 803                 RETURN(0);
 804         }
 805
 806         LDLM_DEBUG(lock, "client-side async enqueue: granted/glimpsed");
 807
 808         stripe = ll_lock_to_stripe_offset(inode, lock);
 809         if (stripe < 0)
 810                 goto iput;
 811
 812         if (lock->l_lvb_len) {
 813                 struct lov_stripe_md *lsm = lli->lli_smd;
 814                 __u64 kms;
 815                 lvb = lock->l_lvb_data;
 816                 lsm->lsm_oinfo[stripe].loi_rss = lvb->lvb_size;
 817
 818                 l_lock(&lock->l_resource->lr_namespace->ns_lock);
 819                 down(&inode->i_sem);
 820                 kms = MAX(lsm->lsm_oinfo[stripe].loi_kms, lvb->lvb_size);
 821                 kms = ldlm_extent_shift_kms(NULL, kms);
 822                 if (lsm->lsm_oinfo[stripe].loi_kms != kms)
 823                         LDLM_DEBUG(lock, "updating kms from "LPU64" to "LPU64,
 824                                    lsm->lsm_oinfo[stripe].loi_kms, kms);
 825                 lsm->lsm_oinfo[stripe].loi_kms = kms;
 826                 up(&inode->i_sem);
 827                 l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 828         }
 829
 830 iput:
 831         iput(inode);
 832         wake_up(&lock->l_waitq);
 833
 834         ldlm_lock2handle(lock, &lockh);
 835         ldlm_lock_decref(&lockh, LCK_PR);
 836         RETURN(0);
 837 }
 838 #endif
 839
 840 static int ll_glimpse_callback(struct ldlm_lock *lock, void *reqp)
 841 {
 842         struct ptlrpc_request *req = reqp;
 843         struct inode *inode = ll_inode_from_lock(lock);
 844         struct ll_inode_info *lli;
 845         struct ost_lvb *lvb;
 846         struct lov_stripe_md *lsm;
 847         int rc, size = sizeof(*lvb), stripe;
 848         ENTRY;
 849
 850         if (inode == NULL)
 851                 GOTO(out, rc = -ELDLM_NO_LOCK_DATA);
 852         lli = ll_i2info(inode);
 853         if (lli == NULL)
 854                 GOTO(iput, rc = -ELDLM_NO_LOCK_DATA);
 855
 856         lsm = lli->lli_smd;
 857         if (lsm == NULL)
 858                 GOTO(iput, rc = -ELDLM_NO_LOCK_DATA);
 859
 860         /* First, find out which stripe index this lock corresponds to. */
 861         stripe = ll_lock_to_stripe_offset(inode, lock);
 862         if (stripe < 0)
 863                 GOTO(iput, rc = -ELDLM_NO_LOCK_DATA);
 864
 865         rc = lustre_pack_reply(req, 1, &size, NULL);
 866         if (rc) {
 867                 CERROR("lustre_pack_reply: %d\n", rc);
 868                 GOTO(iput, rc);
 869         }
 870
 871         lvb = lustre_msg_buf(req->rq_repmsg, 0, sizeof(*lvb));
 872         lvb->lvb_size = lli->lli_smd->lsm_oinfo[stripe].loi_kms;
 873         lvb->lvb_mtime = LTIME_S(inode->i_mtime);
 874         lvb->lvb_atime = LTIME_S(inode->i_atime);
 875         lvb->lvb_ctime = LTIME_S(inode->i_ctime);
 876
 877         LDLM_DEBUG(lock, "i_size: %llu -> stripe number %u -> kms "LPU64,
 878                    inode->i_size, stripe, lvb->lvb_size);
 879         GOTO(iput, 0);
 880  iput:
 881         iput(inode);
 882
 883  out:
 884         /* These errors are normal races, so we don't want to fill the console
 885          * with messages by calling ptlrpc_error() */
 886         if (rc == -ELDLM_NO_LOCK_DATA)
 887                 lustre_pack_reply(req, 0, NULL, NULL);
 888
 889         req->rq_status = rc;
 890         return rc;
 891 }
 892
 893 __u64 lov_merge_size(struct lov_stripe_md *lsm, int kms);
 894 __u64 lov_merge_blocks(struct lov_stripe_md *lsm);
 895 __u64 lov_merge_mtime(struct lov_stripe_md *lsm, __u64 current_time);
 896
 897 /* NB: lov_merge_size will prefer locally cached writes if they extend the
 898  * file (because it prefers KMS over RSS when larger) */
 899 int ll_glimpse_size(struct inode *inode)
 900 {
 901         struct ll_inode_info *lli = ll_i2info(inode);
 902         struct ll_sb_info *sbi = ll_i2sbi(inode);
 903         ldlm_policy_data_t policy = { .l_extent = { 0, OBD_OBJECT_EOF } };
 904         struct lustre_handle lockh = { 0 };
 905         int rc, flags = LDLM_FL_HAS_INTENT;
 906         ENTRY;
 907
 908         CDEBUG(D_DLMTRACE, "Glimpsing inode %lu\n", inode->i_ino);
 909
 910         rc = obd_enqueue(sbi->ll_dt_exp, lli->lli_smd, LDLM_EXTENT, &policy,
 911                          LCK_PR, &flags, ll_extent_lock_callback,
 912                          ldlm_completion_ast, ll_glimpse_callback, inode,
 913                          sizeof(struct ost_lvb), lustre_swab_ost_lvb, &lockh);
 914         if (rc == -ENOENT)
 915                 RETURN(rc);
 916
 917         if (rc != 0) {
 918                 CERROR("obd_enqueue returned rc %d, returning -EIO\n", rc);
 919                 RETURN(rc > 0 ? -EIO : rc);
 920         }
 921
 922         down(&lli->lli_size_sem);
 923         inode->i_size = lov_merge_size(lli->lli_smd, 0);
 924         inode->i_blocks = lov_merge_blocks(lli->lli_smd);
 925         up(&lli->lli_size_sem);
 926
 927         LTIME_S(inode->i_mtime) = lov_merge_mtime(lli->lli_smd,
 928                                                   LTIME_S(inode->i_mtime));
 929
 930         CDEBUG(D_DLMTRACE, "glimpse: size: "LPU64", blocks: "LPU64"\n",
 931                (__u64)inode->i_size, (__u64)inode->i_blocks);
 932
 933         obd_cancel(sbi->ll_dt_exp, lli->lli_smd, LCK_PR, &lockh);
 934         RETURN(rc);
 935 }
 936
 937 void ll_stime_record(struct ll_sb_info *sbi, struct timeval *start,
 938                     struct obd_service_time *stime)
 939 {
 940         struct timeval stop;
 941         do_gettimeofday(&stop);
 942
 943         spin_lock(&sbi->ll_lock);
 944         lprocfs_stime_record(stime, &stop, start);
 945         spin_unlock(&sbi->ll_lock);
 946 }
 947
 948 int ll_extent_lock(struct ll_file_data *fd, struct inode *inode,
 949                    struct lov_stripe_md *lsm, int mode,
 950                    ldlm_policy_data_t *policy, struct lustre_handle *lockh,
 951                    int ast_flags, struct obd_service_time *stime)
 952 {
 953         struct ll_inode_info *lli = ll_i2info(inode);
 954         struct ll_sb_info *sbi = ll_i2sbi(inode);
 955         struct timeval start;
 956         int rc;
 957         ENTRY;
 958
 959         LASSERT(lockh->cookie == 0);
 960
 961         /* XXX phil: can we do this?  won't it screw the file size up? */
 962         if ((fd && (fd->fd_flags & LL_FILE_IGNORE_LOCK)) ||
 963             (sbi->ll_flags & LL_SBI_NOLCK))
 964                 RETURN(0);
 965
 966         CDEBUG(D_DLMTRACE, "Locking inode %lu, start "LPU64" end "LPU64"\n",
 967                inode->i_ino, policy->l_extent.start, policy->l_extent.end);
 968
 969         do_gettimeofday(&start);
 970         rc = obd_enqueue(sbi->ll_dt_exp, lsm, LDLM_EXTENT, policy, mode,
 971                          &ast_flags, ll_extent_lock_callback,
 972                          ldlm_completion_ast, ll_glimpse_callback, inode,
 973                          sizeof(struct ost_lvb), lustre_swab_ost_lvb, lockh);
 974         if (rc > 0)
 975                 rc = -EIO;
 976
 977         ll_stime_record(sbi, &start, stime);
 978
 979         if (policy->l_extent.start == 0 &&
 980             policy->l_extent.end == OBD_OBJECT_EOF) {
 981                 /* vmtruncate()->ll_truncate() first sets the i_size and then
 982                  * the kms under both a DLM lock and the i_sem.  If we don't
 983                  * get the i_sem here we can match the DLM lock and reset
 984                  * i_size from the kms before the truncating path has updated
 985                  * the kms.  generic_file_write can then trust the stale i_size
 986                  * when doing appending writes and effectively cancel the
 987                  * result of the truncate.  Getting the i_sem after the enqueue
 988                  * maintains the DLM -> i_sem acquiry order. */
 989                 down(&lli->lli_size_sem);
 990                 inode->i_size = lov_merge_size(lsm, 1);
 991                 up(&lli->lli_size_sem);
 992         }
 993
 994         if (rc == 0) {
 995                 LTIME_S(inode->i_mtime) =
 996                         lov_merge_mtime(lsm, LTIME_S(inode->i_mtime));
 997         }
 998
 999         RETURN(rc);
1000 }
1001
1002 int ll_extent_unlock(struct ll_file_data *fd, struct inode *inode,
1003                      struct lov_stripe_md *lsm, int mode,
1004                      struct lustre_handle *lockh)
1005 {
1006         struct ll_sb_info *sbi = ll_i2sbi(inode);
1007         int rc;
1008         ENTRY;
1009
1010         /* XXX phil: can we do this?  won't it screw the file size up? */
1011         if ((fd && (fd->fd_flags & LL_FILE_IGNORE_LOCK)) ||
1012             (sbi->ll_flags & LL_SBI_NOLCK))
1013                 RETURN(0);
1014
1015         rc = obd_cancel(sbi->ll_dt_exp, lsm, mode, lockh);
1016
1017         RETURN(rc);
1018 }
1019
1020 static ssize_t ll_file_read(struct file *file, char *buf, size_t count,
1021                             loff_t *ppos)
1022 {
1023         struct inode *inode = file->f_dentry->d_inode;
1024         struct ll_inode_info *lli = ll_i2info(inode);
1025         struct lov_stripe_md *lsm = lli->lli_smd;
1026         struct ll_lock_tree tree;
1027         struct ll_lock_tree_node *node;
1028         int rc;
1029         ssize_t retval;
1030         __u64 kms;
1031         ENTRY;
1032         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p),size="LPSZ",offset=%Ld\n",
1033                inode->i_ino, inode->i_generation, inode, count, *ppos);
1034
1035         /* "If nbyte is 0, read() will return 0 and have no other results."
1036          *                      -- Single Unix Spec */
1037         if (count == 0)
1038                 RETURN(0);
1039
1040         lprocfs_counter_add(ll_i2sbi(inode)->ll_stats, LPROC_LL_READ_BYTES,
1041                             count);
1042
1043         if (!lsm)
1044                 RETURN(0);
1045
1046         node = ll_node_from_inode(inode, *ppos, *ppos  + count - 1,
1047                                   LCK_PR);
1048
1049         tree.lt_fd = file->private_data;
1050
1051         rc = ll_tree_lock(&tree, node, inode, buf, count,
1052                           file->f_flags & O_NONBLOCK ? LDLM_FL_BLOCK_NOWAIT :0);
1053         if (rc != 0)
1054                 RETURN(rc);
1055
1056         down(&lli->lli_size_sem);
1057         kms = lov_merge_size(lsm, 1);
1058         if (*ppos + count - 1 > kms) {
1059                 /* A glimpse is necessary to determine whether we return a short
1060                  * read or some zeroes at the end of the buffer */
1061                 up(&lli->lli_size_sem);
1062                 retval = ll_glimpse_size(inode);
1063                 if (retval)
1064                         goto out;
1065         } else {
1066                 inode->i_size = kms;
1067                 up(&lli->lli_size_sem);
1068         }
1069
1070         CDEBUG(D_INFO, "Read ino %lu, "LPSZ" bytes, offset %lld, i_size %llu\n",
1071                inode->i_ino, count, *ppos, inode->i_size);
1072
1073         /* turn off the kernel's read-ahead */
1074 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
1075         file->f_ramax = 0;
1076 #else
1077         file->f_ra.ra_pages = 0;
1078 #endif
1079         retval = generic_file_read(file, buf, count, ppos);
1080
1081  out:
1082         ll_tree_unlock(&tree, inode);
1083         RETURN(retval);
1084 }
1085
1086 /*
1087  * Write to a file (through the page cache).
1088  */
1089 static ssize_t ll_file_write(struct file *file, const char *buf,
1090                              size_t count, loff_t *ppos)
1091 {
1092         struct inode *inode = file->f_dentry->d_inode;
1093         loff_t maxbytes = ll_file_maxbytes(inode);
1094         struct ll_lock_tree tree;
1095         struct ll_lock_tree_node *node;
1096         ssize_t retval;
1097         int rc;
1098
1099         ENTRY;
1100         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p),size="LPSZ",offset=%Ld\n",
1101                inode->i_ino, inode->i_generation, inode, count, *ppos);
1102
1103         SIGNAL_MASK_ASSERT(); /* XXX BUG 1511 */
1104
1105         /* POSIX, but surprised the VFS doesn't check this already */
1106         if (count == 0)
1107                 RETURN(0);
1108
1109         /* If file was opened for LL_IOC_LOV_SETSTRIPE but the ioctl wasn't
1110          * called on the file, don't fail the below assertion (bug 2388). */
1111         if (file->f_flags & O_LOV_DELAY_CREATE &&
1112             ll_i2info(inode)->lli_smd == NULL)
1113                 RETURN(-EBADF);
1114
1115         LASSERT(ll_i2info(inode)->lli_smd != NULL);
1116
1117         if (file->f_flags & O_APPEND)
1118                 node = ll_node_from_inode(inode, 0, OBD_OBJECT_EOF, LCK_PW);
1119         else
1120                 node = ll_node_from_inode(inode, *ppos, *ppos  + count - 1,
1121                                           LCK_PW);
1122
1123         if (IS_ERR(node))
1124                 RETURN(PTR_ERR(node));
1125
1126         tree.lt_fd = file->private_data;
1127
1128         rc = ll_tree_lock(&tree, node, inode, buf, count,
1129                           file->f_flags & O_NONBLOCK ? LDLM_FL_BLOCK_NOWAIT :0);
1130         if (rc != 0)
1131                 RETURN(rc);
1132
1133         /* this is ok, g_f_w will overwrite this under i_sem if it races
1134          * with a local truncate, it just makes our maxbyte checking easier */
1135         if (file->f_flags & O_APPEND)
1136                 *ppos = inode->i_size;
1137
1138         if (*ppos >= maxbytes) {
1139                 if (count || *ppos > maxbytes) {
1140                         send_sig(SIGXFSZ, current, 0);
1141                         GOTO(out, retval = -EFBIG);
1142                 }
1143         }
1144         if (*ppos + count > maxbytes)
1145                 count = maxbytes - *ppos;
1146
1147         CDEBUG(D_INFO, "Writing inode %lu, "LPSZ" bytes, offset %Lu\n",
1148                inode->i_ino, count, *ppos);
1149
1150         /* generic_file_write handles O_APPEND after getting i_sem */
1151         retval = generic_file_write(file, buf, count, ppos);
1152         EXIT;
1153 out:
1154         ll_tree_unlock(&tree, inode);
1155         /* serialize with mmap/munmap/mremap */
1156         lprocfs_counter_add(ll_i2sbi(inode)->ll_stats, LPROC_LL_WRITE_BYTES,
1157                             retval > 0 ? retval : 0);
1158         return retval;
1159 }
1160
1161 static int ll_lov_setstripe_ea_info(struct inode *inode, struct file *file,
1162                                     int flags, struct lov_user_md *lum,
1163                                     int lum_size)
1164 {
1165         struct ll_inode_info *lli = ll_i2info(inode);
1166         struct file *f;
1167         struct obd_export *exp = ll_i2dtexp(inode);
1168         struct lov_stripe_md *lsm;
1169         struct lookup_intent oit = {.it_op = IT_OPEN, .it_flags = flags};
1170         struct ptlrpc_request *req = NULL;
1171         int rc = 0;
1172         struct lustre_md md;
1173         struct obd_client_handle *och;
1174         ENTRY;
1175
1176
1177         if ((file->f_flags+1) & O_ACCMODE)
1178                 oit.it_flags++;
1179         if (file->f_flags & O_TRUNC)
1180                 oit.it_flags |= 2;
1181
1182         down(&lli->lli_open_sem);
1183         lsm = lli->lli_smd;
1184         if (lsm) {
1185                 up(&lli->lli_open_sem);
1186                 CDEBUG(D_IOCTL, "stripe already exists for ino %lu\n",
1187                        inode->i_ino);
1188                 RETURN(-EEXIST);
1189         }
1190
1191         f = get_empty_filp();
1192         if (!f)
1193                 GOTO(out, -ENOMEM);
1194
1195         f->f_dentry = file->f_dentry;
1196         f->f_vfsmnt = file->f_vfsmnt;
1197         f->f_flags = flags;
1198
1199         rc = ll_intent_alloc(&oit);
1200         if (rc)
1201                 GOTO(out, rc);
1202
1203         rc = ll_intent_file_open(f, lum, lum_size, &oit);
1204         if (rc)
1205                 GOTO(out, rc);
1206         if (it_disposition(&oit, DISP_LOOKUP_NEG))
1207                 GOTO(out, -ENOENT);
1208
1209         req = LUSTRE_IT(&oit)->it_data;
1210         rc = LUSTRE_IT(&oit)->it_status;
1211
1212         if (rc < 0)
1213                 GOTO(out, rc);
1214
1215         rc = mdc_req2lustre_md(ll_i2mdexp(inode), req, 1, exp, &md);
1216         if (rc)
1217                 GOTO(out, rc);
1218         ll_update_inode(f->f_dentry->d_inode, &md);
1219
1220         OBD_ALLOC(och, sizeof(struct obd_client_handle));
1221         rc = ll_local_open(f, &oit, och);
1222         if (rc) { /* Actually ll_local_open cannot fail! */
1223                 GOTO(out, rc);
1224         }
1225         if (LUSTRE_IT(&oit)->it_lock_mode) {
1226                 ldlm_lock_decref_and_cancel((struct lustre_handle *)
1227                                             &LUSTRE_IT(&oit)->it_lock_handle,
1228                                             LUSTRE_IT(&oit)->it_lock_mode);
1229                 LUSTRE_IT(&oit)->it_lock_mode = 0;
1230         }
1231
1232         ll_intent_release(&oit);
1233
1234         /* ll_file_release will decrease the count, but won't free anything
1235            because we have at least one more reference coming from actual open
1236          */
1237         down(&lli->lli_och_sem);
1238         lli->lli_open_fd_write_count++;
1239         up(&lli->lli_och_sem);
1240         rc = ll_file_release(f->f_dentry->d_inode, f);
1241
1242         /* Now also destroy our supplemental och */
1243         ll_md_och_close(ll_i2mdexp(inode), f->f_dentry->d_inode, och);
1244         EXIT;
1245  out:
1246         ll_intent_release(&oit);
1247         if (f)
1248                 put_filp(f);
1249         up(&lli->lli_open_sem);
1250         if (req != NULL)
1251                 ptlrpc_req_finished(req);
1252         return rc;
1253 }
1254
1255 static int ll_lov_setea(struct inode *inode, struct file *file,
1256                         unsigned long arg)
1257 {
1258         int flags = MDS_OPEN_HAS_OBJS | FMODE_WRITE;
1259         struct lov_user_md  *lump;
1260         int lum_size = sizeof(struct lov_user_md) +
1261                 sizeof(struct lov_user_ost_data);
1262         int rc;
1263         ENTRY;
1264
1265         if (!capable (CAP_SYS_ADMIN))
1266                 RETURN(-EPERM);
1267
1268         OBD_ALLOC(lump, lum_size);
1269         if (lump == NULL) {
1270                 RETURN(-ENOMEM);
1271         }
1272         rc = copy_from_user(lump, (struct lov_user_md  *)arg, lum_size);
1273         if (rc) {
1274                 OBD_FREE(lump, lum_size);
1275                 RETURN(-EFAULT);
1276         }
1277
1278         rc = ll_lov_setstripe_ea_info(inode, file, flags, lump, lum_size);
1279
1280         OBD_FREE(lump, lum_size);
1281         RETURN(rc);
1282 }
1283
1284 static int ll_lov_setstripe(struct inode *inode, struct file *file,
1285                             unsigned long arg)
1286 {
1287         struct lov_user_md lum, *lump = (struct lov_user_md *)arg;
1288         int rc;
1289         int flags = FMODE_WRITE;
1290         ENTRY;
1291
1292         /* Bug 1152: copy properly when this is no longer true */
1293         LASSERT(sizeof(lum) == sizeof(*lump));
1294         LASSERT(sizeof(lum.lmm_objects[0]) == sizeof(lump->lmm_objects[0]));
1295         rc = copy_from_user(&lum, lump, sizeof(lum));
1296         if (rc)
1297                 RETURN(-EFAULT);
1298
1299         rc = ll_lov_setstripe_ea_info(inode, file, flags, &lum, sizeof(lum));
1300         if (rc == 0) {
1301                  put_user(0, &lump->lmm_stripe_count);
1302                  rc = obd_iocontrol(LL_IOC_LOV_GETSTRIPE, ll_i2dtexp(inode),
1303                                     0, ll_i2info(inode)->lli_smd, lump);
1304         }
1305         RETURN(rc);
1306 }
1307
1308 static int ll_lov_getstripe(struct inode *inode, unsigned long arg)
1309 {
1310         struct lov_stripe_md *lsm = ll_i2info(inode)->lli_smd;
1311
1312         if (!lsm)
1313                 RETURN(-ENODATA);
1314
1315         return obd_iocontrol(LL_IOC_LOV_GETSTRIPE, ll_i2dtexp(inode), 0, lsm,
1316                              (void *)arg);
1317 }
1318
1319 static int ll_get_grouplock(struct inode *inode, struct file *file,
1320                          unsigned long arg)
1321 {
1322         struct ll_file_data *fd = file->private_data;
1323         ldlm_policy_data_t policy = { .l_extent = { .start = 0,
1324                                                     .end = OBD_OBJECT_EOF}};
1325         struct lustre_handle lockh = { 0 };
1326         struct ll_inode_info *lli = ll_i2info(inode);
1327         struct lov_stripe_md *lsm = lli->lli_smd;
1328         int flags = 0, rc;
1329         ENTRY;
1330
1331         if (fd->fd_flags & LL_FILE_GROUP_LOCKED) {
1332                 RETURN(-EINVAL);
1333         }
1334
1335         policy.l_extent.gid = arg;
1336         if (file->f_flags & O_NONBLOCK)
1337                 flags = LDLM_FL_BLOCK_NOWAIT;
1338
1339         rc = ll_extent_lock(fd, inode, lsm, LCK_GROUP, &policy, &lockh, flags,
1340                             &ll_i2sbi(inode)->ll_grouplock_stime);
1341         if (rc != 0)
1342                 RETURN(rc);
1343
1344         fd->fd_flags |= LL_FILE_GROUP_LOCKED|LL_FILE_IGNORE_LOCK;
1345         fd->fd_gid = arg;
1346         memcpy(&fd->fd_cwlockh, &lockh, sizeof(lockh));
1347
1348         RETURN(0);
1349 }
1350
1351 static int ll_put_grouplock(struct inode *inode, struct file *file,
1352                          unsigned long arg)
1353 {
1354         struct ll_file_data *fd = file->private_data;
1355         struct ll_inode_info *lli = ll_i2info(inode);
1356         struct lov_stripe_md *lsm = lli->lli_smd;
1357         int rc;
1358         ENTRY;
1359
1360         if (!(fd->fd_flags & LL_FILE_GROUP_LOCKED)) {
1361                 /* Ugh, it's already unlocked. */
1362                 RETURN(-EINVAL);
1363         }
1364
1365         if (fd->fd_gid != arg) /* Ugh? Unlocking with different gid? */
1366                 RETURN(-EINVAL);
1367
1368         fd->fd_flags &= ~(LL_FILE_GROUP_LOCKED|LL_FILE_IGNORE_LOCK);
1369
1370         rc = ll_extent_unlock(fd, inode, lsm, LCK_GROUP, &fd->fd_cwlockh);
1371         if (rc)
1372                 RETURN(rc);
1373
1374         fd->fd_gid = 0;
1375         memset(&fd->fd_cwlockh, 0, sizeof(fd->fd_cwlockh));
1376
1377         RETURN(0);
1378 }
1379
1380 int ll_file_ioctl(struct inode *inode, struct file *file, unsigned int cmd,
1381                   unsigned long arg)
1382 {
1383         struct ll_file_data *fd = file->private_data;
1384         struct ll_sb_info *sbi = ll_i2sbi(inode);
1385         int flags;
1386         ENTRY;
1387
1388         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p),cmd=%x\n", inode->i_ino,
1389                inode->i_generation, inode, cmd);
1390
1391         if (_IOC_TYPE(cmd) == 'T') /* tty ioctls */
1392                 RETURN(-ENOTTY);
1393
1394         lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats, LPROC_LL_IOCTL);
1395         switch(cmd) {
1396         case LL_IOC_GETFLAGS:
1397                 /* Get the current value of the file flags */
1398                 return put_user(fd->fd_flags, (int *)arg);
1399         case LL_IOC_SETFLAGS:
1400         case LL_IOC_CLRFLAGS:
1401                 /* Set or clear specific file flags */
1402                 /* XXX This probably needs checks to ensure the flags are
1403                  *     not abused, and to handle any flag side effects.
1404                  */
1405                 if (get_user(flags, (int *) arg))
1406                         RETURN(-EFAULT);
1407
1408                 if (cmd == LL_IOC_SETFLAGS)
1409                         fd->fd_flags |= flags;
1410                 else
1411                         fd->fd_flags &= ~flags;
1412                 RETURN(0);
1413         case LL_IOC_LOV_SETSTRIPE:
1414                 RETURN(ll_lov_setstripe(inode, file, arg));
1415         case LL_IOC_LOV_SETEA:
1416                 RETURN(ll_lov_setea(inode, file, arg));
1417         case IOC_MDC_SHOWFID: {
1418                 struct lustre_id *idp = (struct lustre_id *)arg;
1419                 struct lustre_id id;
1420                 char *filename;
1421                 int rc;
1422
1423                 filename = getname((const char *)arg);
1424                 if (IS_ERR(filename))
1425                         RETURN(PTR_ERR(filename));
1426
1427                 ll_inode2id(&id, inode);
1428
1429                 rc = ll_get_fid(sbi->ll_md_exp, &id, filename, &id);
1430                 if (rc < 0)
1431                         GOTO(out_filename, rc);
1432
1433                 rc = copy_to_user(idp, &id, sizeof(*idp));
1434                 if (rc)
1435                         GOTO(out_filename, rc = -EFAULT);
1436
1437                 EXIT;
1438         out_filename:
1439                 putname(filename);
1440                 return rc;
1441         }
1442         case LL_IOC_LOV_GETSTRIPE:
1443                 RETURN(ll_lov_getstripe(inode, arg));
1444         case EXT3_IOC_GETFLAGS:
1445         case EXT3_IOC_SETFLAGS:
1446                 RETURN( ll_iocontrol(inode, file, cmd, arg) );
1447         case LL_IOC_GROUP_LOCK:
1448                 RETURN(ll_get_grouplock(inode, file, arg));
1449         case LL_IOC_GROUP_UNLOCK:
1450                 RETURN(ll_put_grouplock(inode, file, arg));
1451         case EXT3_IOC_GETVERSION_OLD:
1452         case EXT3_IOC_GETVERSION:
1453                 return put_user(inode->i_generation, (int *) arg);
1454         /* We need to special case any other ioctls we want to handle,
1455          * to send them to the MDS/OST as appropriate and to properly
1456          * network encode the arg field.
1457         case EXT2_IOC_GETVERSION_OLD:
1458         case EXT2_IOC_GETVERSION_NEW:
1459         case EXT2_IOC_SETVERSION_OLD:
1460         case EXT2_IOC_SETVERSION_NEW:
1461         case EXT3_IOC_SETVERSION_OLD:
1462         case EXT3_IOC_SETVERSION:
1463         */
1464         default:
1465                 RETURN( obd_iocontrol(cmd, ll_i2dtexp(inode), 0, NULL,
1466                                       (void *)arg) );
1467         }
1468 }
1469
1470 loff_t ll_file_seek(struct file *file, loff_t offset, int origin)
1471 {
1472         struct inode *inode = file->f_dentry->d_inode;
1473         struct ll_file_data *fd = file->private_data;
1474         struct lov_stripe_md *lsm = ll_i2info(inode)->lli_smd;
1475         struct lustre_handle lockh = {0};
1476         loff_t retval;
1477         ENTRY;
1478         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p),to=%llu\n", inode->i_ino,
1479                inode->i_generation, inode,
1480                offset + ((origin==2) ? inode->i_size : file->f_pos));
1481
1482         lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats, LPROC_LL_LLSEEK);
1483         if (origin == 2) { /* SEEK_END */
1484                 ldlm_policy_data_t policy = { .l_extent = {0, OBD_OBJECT_EOF }};
1485                 struct ll_inode_info *lli = ll_i2info(inode);
1486                 int nonblock = 0, rc;
1487
1488                 if (file->f_flags & O_NONBLOCK)
1489                         nonblock = LDLM_FL_BLOCK_NOWAIT;
1490
1491                 rc = ll_extent_lock(fd, inode, lsm, LCK_PR, &policy, &lockh,
1492                                     nonblock, &ll_i2sbi(inode)->ll_seek_stime);
1493                 if (rc != 0)
1494                         RETURN(rc);
1495
1496                 down(&lli->lli_size_sem);
1497                 offset += inode->i_size;
1498                 up(&lli->lli_size_sem);
1499         } else if (origin == 1) { /* SEEK_CUR */
1500                 offset += file->f_pos;
1501         }
1502
1503         retval = -EINVAL;
1504         if (offset >= 0 && offset <= ll_file_maxbytes(inode)) {
1505                 if (offset != file->f_pos) {
1506                         file->f_pos = offset;
1507 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
1508                         file->f_reada = 0;
1509                         file->f_version = ++event;
1510 #endif
1511                 }
1512                 retval = offset;
1513         }
1514
1515         if (origin == 2)
1516                 ll_extent_unlock(fd, inode, lsm, LCK_PR, &lockh);
1517         RETURN(retval);
1518 }
1519
1520 int ll_fsync(struct file *file, struct dentry *dentry, int data)
1521 {
1522         struct inode *inode = dentry->d_inode;
1523         struct lov_stripe_md *lsm = ll_i2info(inode)->lli_smd;
1524         struct lustre_id id;
1525         struct ptlrpc_request *req;
1526         int rc, err;
1527         ENTRY;
1528         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p)\n", inode->i_ino,
1529                inode->i_generation, inode);
1530
1531         lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats, LPROC_LL_FSYNC);
1532
1533         /* fsync's caller has already called _fdata{sync,write}, we want
1534          * that IO to finish before calling the osc and mdc sync methods */
1535         rc = filemap_fdatawait(inode->i_mapping);
1536
1537         ll_inode2id(&id, inode);
1538         err = md_sync(ll_i2sbi(inode)->ll_md_exp, &id, &req);
1539         if (!rc)
1540                 rc = err;
1541         if (!err)
1542                 ptlrpc_req_finished(req);
1543
1544         if (data && lsm) {
1545                 struct obdo *oa = obdo_alloc();
1546
1547                 if (!oa)
1548                         RETURN(rc ? rc : -ENOMEM);
1549
1550                 oa->o_id = lsm->lsm_object_id;
1551                 oa->o_gr = lsm->lsm_object_gr;
1552                 oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
1553
1554                 obdo_from_inode(oa, inode, (OBD_MD_FLTYPE | OBD_MD_FLATIME |
1555                                             OBD_MD_FLMTIME | OBD_MD_FLCTIME |
1556                                             OBD_MD_FLGROUP));
1557
1558                 err = obd_sync(ll_i2sbi(inode)->ll_dt_exp, oa, lsm,
1559                                0, OBD_OBJECT_EOF);
1560                 if (!rc)
1561                         rc = err;
1562                 obdo_free(oa);
1563         }
1564
1565         RETURN(rc);
1566 }
1567
1568 int ll_file_flock(struct file *file, int cmd, struct file_lock *file_lock)
1569 {
1570         struct inode *inode = file->f_dentry->d_inode;
1571         struct ll_inode_info *li = ll_i2info(inode);
1572         struct ll_sb_info *sbi = ll_i2sbi(inode);
1573         struct obd_device *obddev;
1574         struct ldlm_res_id res_id =
1575                 { .name = {id_fid(&li->lli_id), id_group(&li->lli_id), LDLM_FLOCK} };
1576         struct lustre_handle lockh = {0};
1577         ldlm_policy_data_t flock;
1578         ldlm_mode_t mode = 0;
1579         int flags = 0;
1580         int rc;
1581         ENTRY;
1582
1583         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu file_lock=%p\n",
1584                inode->i_ino, file_lock);
1585
1586         flock.l_flock.pid = file_lock->fl_pid;
1587         flock.l_flock.start = file_lock->fl_start;
1588         flock.l_flock.end = file_lock->fl_end;
1589
1590         switch (file_lock->fl_type) {
1591         case F_RDLCK:
1592                 mode = LCK_PR;
1593                 break;
1594         case F_UNLCK:
1595                 /* An unlock request may or may not have any relation to
1596                  * existing locks so we may not be able to pass a lock handle
1597                  * via a normal ldlm_lock_cancel() request. The request may even
1598                  * unlock a byte range in the middle of an existing lock. In
1599                  * order to process an unlock request we need all of the same
1600                  * information that is given with a normal read or write record
1601                  * lock request. To avoid creating another ldlm unlock (cancel)
1602                  * message we'll treat a LCK_NL flock request as an unlock. */
1603                 mode = LCK_NL;
1604                 break;
1605         case F_WRLCK:
1606                 mode = LCK_PW;
1607                 break;
1608         default:
1609                 CERROR("unknown fcntl lock type: %d\n", file_lock->fl_type);
1610                 LBUG();
1611         }
1612
1613         switch (cmd) {
1614         case F_SETLKW:
1615 #ifdef F_SETLKW64
1616         case F_SETLKW64:
1617 #endif
1618                 flags = 0;
1619                 break;
1620         case F_SETLK:
1621 #ifdef F_SETLK64
1622         case F_SETLK64:
1623 #endif
1624                 flags = LDLM_FL_BLOCK_NOWAIT;
1625                 break;
1626         case F_GETLK:
1627 #ifdef F_GETLK64
1628         case F_GETLK64:
1629 #endif
1630                 flags = LDLM_FL_TEST_LOCK;
1631                 /* Save the old mode so that if the mode in the lock changes we
1632                  * can decrement the appropriate reader or writer refcount. */
1633                 file_lock->fl_type = mode;
1634                 break;
1635         default:
1636                 CERROR("unknown fcntl lock command: %d\n", cmd);
1637                 LBUG();
1638         }
1639
1640         CDEBUG(D_DLMTRACE, "inode=%lu, pid="LPU64", flags=%#x, mode=%u, "
1641                "start="LPU64", end="LPU64"\n", inode->i_ino, flock.l_flock.pid,
1642                flags, mode, flock.l_flock.start, flock.l_flock.end);
1643
1644         obddev = md_get_real_obd(sbi->ll_md_exp, &li->lli_id);
1645         rc = ldlm_cli_enqueue(obddev->obd_self_export, NULL,
1646                               obddev->obd_namespace,
1647                               res_id, LDLM_FLOCK, &flock, mode, &flags,
1648                               NULL, ldlm_flock_completion_ast, NULL, file_lock,
1649                               NULL, 0, NULL, &lockh);
1650         RETURN(rc);
1651 }
1652
1653 int ll_inode_revalidate_it(struct dentry *dentry)
1654 {
1655         struct lookup_intent oit = { .it_op = IT_GETATTR };
1656         struct inode *inode = dentry->d_inode;
1657         struct ptlrpc_request *req = NULL;
1658         struct ll_inode_info *lli;
1659         struct lov_stripe_md *lsm;
1660         struct ll_sb_info *sbi;
1661         struct lustre_id id;
1662         int rc;
1663         ENTRY;
1664
1665         if (!inode) {
1666                 CERROR("REPORT THIS LINE TO PETER\n");
1667                 RETURN(0);
1668         }
1669
1670         sbi = ll_i2sbi(inode);
1671
1672         ll_inode2id(&id, inode);
1673         lli = ll_i2info(inode);
1674         LASSERT(id_fid(&id) != 0);
1675
1676         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p), name=%s(%p)\n",
1677                inode->i_ino, inode->i_generation, inode, dentry->d_name.name,
1678                dentry);
1679
1680 #if (LINUX_VERSION_CODE <= KERNEL_VERSION(2,5,0))
1681         lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats, LPROC_LL_REVALIDATE);
1682 #endif
1683
1684         rc = ll_intent_alloc(&oit);
1685         if (rc)
1686                 RETURN(-ENOMEM);
1687
1688         rc = md_intent_lock(sbi->ll_md_exp, &id, NULL, 0, NULL, 0, &id,
1689                             &oit, 0, &req, ll_mdc_blocking_ast);
1690         if (rc < 0)
1691                 GOTO(out, rc);
1692
1693         rc = revalidate_it_finish(req, 1, &oit, dentry);
1694         if (rc) {
1695                 GOTO(out, rc);
1696         }
1697
1698         ll_lookup_finish_locks(&oit, dentry);
1699
1700         lsm = lli->lli_smd;
1701         if (lsm == NULL) /* object not yet allocated, don't validate size */
1702                 GOTO(out, rc = 0);
1703
1704         /*
1705          * ll_glimpse_size() will prefer locally cached writes if they extend
1706          * the file.
1707          */
1708         rc = ll_glimpse_size(inode);
1709         EXIT;
1710 out:
1711         ll_intent_release(&oit);
1712         if (req)
1713                 ptlrpc_req_finished(req);
1714         return rc;
1715 }
1716
1717 #if (LINUX_VERSION_CODE > KERNEL_VERSION(2,5,0))
1718 int ll_getattr(struct vfsmount *mnt, struct dentry *de, struct kstat *stat)
1719 {
1720         int res = 0;
1721         struct inode *inode = de->d_inode;
1722         struct ll_inode_info *lli = ll_i2info(inode);
1723
1724         res = ll_inode_revalidate_it(de);
1725         lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats, LPROC_LL_GETATTR);
1726
1727         if (res)
1728                 return res;
1729
1730         stat->ino = inode->i_ino;
1731         stat->mode = inode->i_mode;
1732         stat->nlink = inode->i_nlink;
1733         stat->uid = inode->i_uid;
1734         stat->gid = inode->i_gid;
1735         stat->atime = inode->i_atime;
1736         stat->mtime = inode->i_mtime;
1737         stat->ctime = inode->i_ctime;
1738         stat->blksize = inode->i_blksize;
1739
1740         down(&lli->lli_size_sem);
1741         stat->size = inode->i_size;
1742         stat->blocks = inode->i_blocks;
1743         up(&lli->lli_size_sem);
1744
1745         stat->rdev = kdev_t_to_nr(inode->i_rdev);
1746         stat->dev = id_group(&ll_i2info(inode)->lli_id);
1747         return 0;
1748 }
1749 #endif
1750
1751 static
1752 int ll_setxattr_internal(struct inode *inode, const char *name,
1753                          const void *value, size_t size, int flags,
1754                          __u64 valid)
1755 {
1756         struct ll_sb_info *sbi = ll_i2sbi(inode);
1757         struct ptlrpc_request *request = NULL;
1758         struct mdc_op_data op_data;
1759         struct iattr attr;
1760         int rc = 0;
1761         ENTRY;
1762
1763         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu\n", inode->i_ino);
1764         lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats, LPROC_LL_SETXATTR);
1765
1766         memset(&attr, 0x0, sizeof(attr));
1767         attr.ia_valid |= valid;
1768         attr.ia_attr_flags = flags;
1769
1770         ll_prepare_mdc_data(&op_data, inode, NULL, NULL, 0, 0);
1771
1772         rc = md_setattr(sbi->ll_md_exp, &op_data, &attr,
1773                         (void*) name, strnlen(name, XATTR_NAME_MAX)+1,
1774                         (void*) value, size, &request);
1775         if (rc) {
1776                 CERROR("md_setattr fails: rc = %d\n", rc);
1777                 GOTO(out, rc);
1778         }
1779
1780  out:
1781         ptlrpc_req_finished(request);
1782         RETURN(rc);
1783 }
1784
1785 int ll_setxattr(struct dentry *dentry, const char *name, const void *value,
1786                 size_t size, int flags)
1787 {
1788         int rc, error;
1789         struct posix_acl *acl;
1790         struct ll_inode_info *lli;
1791         ENTRY;
1792
1793         rc = ll_setxattr_internal(dentry->d_inode, name, value, size,
1794                                   flags, ATTR_EA);
1795
1796         /* update inode's acl info */
1797         if (rc == 0 && strcmp(name, XATTR_NAME_ACL_ACCESS) == 0) {
1798                 if (value) {
1799                         acl = posix_acl_from_xattr(value, size);
1800                         if (IS_ERR(acl)) {
1801                                 CERROR("convert from xattr to acl error: %ld",
1802                                         PTR_ERR(acl));
1803                                 GOTO(out, rc);
1804                         } else if (acl) {
1805                                 error = posix_acl_valid(acl);
1806                                 if (error) {
1807                                         CERROR("acl valid error: %d", error);
1808                                         posix_acl_release(acl);
1809                                         GOTO(out, rc);
1810                                 }
1811                         }
1812                 } else {
1813                         acl = NULL;
1814                 }
1815
1816                 lli = ll_i2info(dentry->d_inode);
1817                 spin_lock(&lli->lli_lock);
1818                 if (lli->lli_acl_access != NULL)
1819                         posix_acl_release(lli->lli_acl_access);
1820                 lli->lli_acl_access = acl;
1821                 spin_unlock(&lli->lli_lock);
1822         }
1823         EXIT;
1824 out:
1825         return(rc);
1826 }
1827
1828 int ll_removexattr(struct dentry *dentry, const char *name)
1829 {
1830         return ll_setxattr_internal(dentry->d_inode, name, NULL, 0, 0,
1831                                     ATTR_EA_RM);
1832 }
1833
1834 static
1835 int ll_getxattr_internal(struct inode *inode, const char *name, int namelen,
1836                          void *value, size_t size, __u64 valid)
1837 {
1838         struct ptlrpc_request *request = NULL;
1839         struct ll_sb_info *sbi = ll_i2sbi(inode);
1840         struct lustre_id id;
1841         struct mds_body *body;
1842         void *ea_data;
1843         int rc, ea_size;
1844         ENTRY;
1845
1846         lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats, LPROC_LL_GETXATTR);
1847
1848         ll_inode2id(&id, inode);
1849         rc = md_getattr(sbi->ll_md_exp, &id, valid, name, namelen,
1850                          size, &request);
1851         if (rc) {
1852                 if (rc != -ENODATA && rc != -EOPNOTSUPP)
1853                         CERROR("md_getattr fails: rc = %d\n", rc);
1854                 GOTO(out, rc);
1855         }
1856
1857         body = lustre_msg_buf(request->rq_repmsg, 0, sizeof(*body));
1858         LASSERT(body != NULL);
1859         LASSERT_REPSWABBED(request, 0);
1860
1861         ea_size = body->eadatasize;
1862         LASSERT(ea_size <= request->rq_repmsg->buflens[0]);
1863
1864         if (size == 0)
1865                 GOTO(out, rc = ea_size);
1866
1867         ea_data = lustre_msg_buf(request->rq_repmsg, 1, ea_size);
1868         LASSERT(ea_data != NULL);
1869         LASSERT_REPSWABBED(request, 1);
1870
1871         if (value)
1872                 memcpy(value, ea_data, ea_size);
1873         rc = ea_size;
1874  out:
1875         ptlrpc_req_finished(request);
1876         RETURN(rc);
1877 }
1878
1879 int ll_getxattr(struct dentry *dentry, const char *name, void *value,
1880                 size_t size)
1881 {
1882         return ll_getxattr_internal(dentry->d_inode, name, strlen(name) + 1,
1883                                     value, size, OBD_MD_FLEA);
1884 }
1885
1886 int ll_listxattr(struct dentry *dentry, char *list, size_t size)
1887 {
1888         return ll_getxattr_internal(dentry->d_inode, NULL, 0, list, size,
1889                                     OBD_MD_FLEALIST);
1890 }
1891
1892 /*
1893  * XXX We could choose not to check DLM lock. Leave the decision
1894  * to remote acl handling.
1895  */
1896 static int
1897 lustre_check_acl(struct inode *inode, int mask)
1898 {
1899         struct lookup_intent it = { .it_op = IT_GETATTR };
1900         struct dentry de = { .d_inode = inode };
1901         struct ll_sb_info *sbi;
1902         struct lustre_id id;
1903         struct ptlrpc_request *req = NULL;
1904         struct ll_inode_info *lli = ll_i2info(inode);
1905         struct posix_acl *acl;
1906         int rc;
1907         ENTRY;
1908
1909         sbi = ll_i2sbi(inode);
1910         ll_inode2id(&id, inode);
1911
1912         if (ll_intent_alloc(&it))
1913                 return -EACCES;
1914
1915         rc = md_intent_lock(sbi->ll_md_exp, &id, NULL, 0, NULL, 0, &id,
1916                             &it, 0, &req, ll_mdc_blocking_ast);
1917         if (rc < 0) {
1918                 ll_intent_free(&it);
1919                 GOTO(out, rc);
1920         }
1921
1922         rc = revalidate_it_finish(req, 1, &it, &de);
1923         if (rc) {
1924                 ll_intent_release(&it);
1925                 GOTO(out, rc);
1926         }
1927
1928         ll_lookup_finish_locks(&it, &de);
1929         ll_intent_free(&it);
1930
1931         spin_lock(&lli->lli_lock);
1932         acl = posix_acl_dup(ll_i2info(inode)->lli_acl_access);
1933         spin_unlock(&lli->lli_lock);
1934
1935         if (!acl)
1936                 GOTO(out, rc = -EAGAIN);
1937
1938         rc = posix_acl_permission(inode, acl, mask);
1939         posix_acl_release(acl);
1940
1941 out:
1942         if (req)
1943                 ptlrpc_req_finished(req);
1944
1945         RETURN(rc);
1946 }
1947
1948 int ll_inode_permission(struct inode *inode, int mask, struct nameidata *nd)
1949 {
1950         return generic_permission(inode, mask, lustre_check_acl);
1951 }
1952
1953 struct file_operations ll_file_operations = {
1954         .read           = ll_file_read,
1955         .write          = ll_file_write,
1956         .ioctl          = ll_file_ioctl,
1957         .open           = ll_file_open,
1958         .release        = ll_file_release,
1959         .mmap           = ll_file_mmap,
1960         .llseek         = ll_file_seek,
1961 #if (LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,0))
1962         .sendfile       = generic_file_sendfile,
1963 #endif
1964         .fsync          = ll_fsync,
1965         .lock           = ll_file_flock
1966 };
1967
1968 struct inode_operations ll_file_inode_operations = {
1969         .setattr        = ll_setattr,
1970         .truncate       = ll_truncate,
1971 #if (LINUX_VERSION_CODE > KERNEL_VERSION(2,5,0))
1972         .getattr        = ll_getattr,
1973 #else
1974         .revalidate_it  = ll_inode_revalidate_it,
1975 #endif
1976         .setxattr       = ll_setxattr,
1977         .getxattr       = ll_getxattr,
1978         .listxattr      = ll_listxattr,
1979         .removexattr    = ll_removexattr,
1980         .permission     = ll_inode_permission,
1981 };
1982