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