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         struct dentry *parent = file->f_dentry->d_parent;
 239         const char *name = file->f_dentry->d_name.name;
 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_recreate_obj(struct inode *inode, struct file *file,
1162                                unsigned long arg)
1163 {
1164         struct ll_inode_info *lli = ll_i2info(inode);
1165         struct obd_export *exp = ll_i2dtexp(inode);
1166         struct ll_recreate_obj ucreatp;
1167         struct obd_trans_info oti = { 0 };
1168         struct obdo *oa = NULL;
1169         int lsm_size;
1170         int rc = 0;
1171         struct lov_stripe_md *lsm, *lsm2;
1172         ENTRY;
1173
1174         if (!capable (CAP_SYS_ADMIN))
1175                 RETURN(-EPERM);
1176
1177         rc = copy_from_user(&ucreatp, (struct ll_recreate_obj *)arg,
1178                             sizeof(struct ll_recreate_obj));
1179         if (rc) {
1180                 RETURN(-EFAULT);
1181         }
1182         oa = obdo_alloc();
1183         if (oa == NULL)
1184                 RETURN(-ENOMEM);
1185
1186         down(&lli->lli_open_sem);
1187         lsm = lli->lli_smd;
1188         if (lsm == NULL)
1189                 GOTO(out, rc = -ENOENT);
1190         lsm_size = sizeof(*lsm) + (sizeof(struct lov_oinfo) *
1191                    (lsm->lsm_stripe_count));
1192
1193         OBD_ALLOC(lsm2, lsm_size);
1194         if (lsm2 == NULL)
1195                 GOTO(out, rc = -ENOMEM);
1196
1197         oa->o_id = ucreatp.lrc_id;
1198         oa->o_nlink = ucreatp.lrc_ost_idx;
1199         oa->o_gr = ucreatp.lrc_group;
1200         oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP | OBD_MD_FLFLAGS;
1201         oa->o_flags |= OBD_FL_RECREATE_OBJS;
1202         obdo_from_inode(oa, inode, OBD_MD_FLTYPE | OBD_MD_FLATIME |
1203                         OBD_MD_FLMTIME | OBD_MD_FLCTIME);
1204
1205         oti.oti_objid = NULL;
1206         memcpy(lsm2, lsm, lsm_size);
1207         rc = obd_create(exp, oa, NULL, 0, &lsm2, &oti);
1208
1209         OBD_FREE(lsm2, lsm_size);
1210         GOTO(out, rc);
1211 out:
1212         up(&lli->lli_open_sem);
1213         obdo_free(oa);
1214         return rc;
1215 }
1216
1217 static int ll_lov_setstripe_ea_info(struct inode *inode, struct file *file,
1218                                     int flags, struct lov_user_md *lum,
1219                                     int lum_size)
1220 {
1221         struct ll_inode_info *lli = ll_i2info(inode);
1222         struct file *f;
1223         struct obd_export *exp = ll_i2dtexp(inode);
1224         struct lov_stripe_md *lsm;
1225         struct lookup_intent oit = {.it_op = IT_OPEN, .it_flags = flags};
1226         struct ptlrpc_request *req = NULL;
1227         int rc = 0;
1228         struct lustre_md md;
1229         struct obd_client_handle *och;
1230         ENTRY;
1231
1232
1233         if ((file->f_flags+1) & O_ACCMODE)
1234                 oit.it_flags++;
1235         if (file->f_flags & O_TRUNC)
1236                 oit.it_flags |= 2;
1237
1238         down(&lli->lli_open_sem);
1239         lsm = lli->lli_smd;
1240         if (lsm) {
1241                 up(&lli->lli_open_sem);
1242                 CDEBUG(D_IOCTL, "stripe already exists for ino %lu\n",
1243                        inode->i_ino);
1244                 RETURN(-EEXIST);
1245         }
1246
1247         f = get_empty_filp();
1248         if (!f)
1249                 GOTO(out, -ENOMEM);
1250
1251         f->f_dentry = file->f_dentry;
1252         f->f_vfsmnt = file->f_vfsmnt;
1253         f->f_flags = flags;
1254
1255         rc = ll_intent_alloc(&oit);
1256         if (rc)
1257                 GOTO(out, rc);
1258
1259         rc = ll_intent_file_open(f, lum, lum_size, &oit);
1260         if (rc)
1261                 GOTO(out, rc);
1262         if (it_disposition(&oit, DISP_LOOKUP_NEG))
1263                 GOTO(out, -ENOENT);
1264
1265         req = LUSTRE_IT(&oit)->it_data;
1266         rc = LUSTRE_IT(&oit)->it_status;
1267
1268         if (rc < 0)
1269                 GOTO(out, rc);
1270
1271         rc = mdc_req2lustre_md(ll_i2mdexp(inode), req, 1, exp, &md);
1272         if (rc)
1273                 GOTO(out, rc);
1274         ll_update_inode(f->f_dentry->d_inode, &md);
1275
1276         OBD_ALLOC(och, sizeof(struct obd_client_handle));
1277         rc = ll_local_open(f, &oit, och);
1278         if (rc) { /* Actually ll_local_open cannot fail! */
1279                 GOTO(out, rc);
1280         }
1281         if (LUSTRE_IT(&oit)->it_lock_mode) {
1282                 ldlm_lock_decref_and_cancel((struct lustre_handle *)
1283                                             &LUSTRE_IT(&oit)->it_lock_handle,
1284                                             LUSTRE_IT(&oit)->it_lock_mode);
1285                 LUSTRE_IT(&oit)->it_lock_mode = 0;
1286         }
1287
1288         ll_intent_release(&oit);
1289
1290         /* ll_file_release will decrease the count, but won't free anything
1291            because we have at least one more reference coming from actual open
1292          */
1293         down(&lli->lli_och_sem);
1294         lli->lli_open_fd_write_count++;
1295         up(&lli->lli_och_sem);
1296         rc = ll_file_release(f->f_dentry->d_inode, f);
1297
1298         /* Now also destroy our supplemental och */
1299         ll_md_och_close(ll_i2mdexp(inode), f->f_dentry->d_inode, och);
1300         EXIT;
1301  out:
1302         ll_intent_release(&oit);
1303         if (f)
1304                 put_filp(f);
1305         up(&lli->lli_open_sem);
1306         if (req != NULL)
1307                 ptlrpc_req_finished(req);
1308         return rc;
1309 }
1310
1311 static int ll_lov_setea(struct inode *inode, struct file *file,
1312                             unsigned long arg)
1313 {
1314         int flags = MDS_OPEN_HAS_OBJS | FMODE_WRITE;
1315         struct lov_user_md  *lump;
1316         int lum_size = sizeof(struct lov_user_md) +
1317                        sizeof(struct lov_user_ost_data);
1318         int rc;
1319         ENTRY;
1320
1321         if (!capable (CAP_SYS_ADMIN))
1322                 RETURN(-EPERM);
1323
1324         OBD_ALLOC(lump, lum_size);
1325         if (lump == NULL) {
1326                 RETURN(-ENOMEM);
1327         }
1328         rc = copy_from_user(lump, (struct lov_user_md  *)arg, lum_size);
1329         if (rc) {
1330                 OBD_FREE(lump, lum_size);
1331                 RETURN(-EFAULT);
1332         }
1333
1334         rc = ll_lov_setstripe_ea_info(inode, file, flags, lump, lum_size);
1335
1336         OBD_FREE(lump, lum_size);
1337         RETURN(rc);
1338 }
1339
1340 static int ll_lov_setstripe(struct inode *inode, struct file *file,
1341                             unsigned long arg)
1342 {
1343         struct lov_user_md lum, *lump = (struct lov_user_md *)arg;
1344         int rc;
1345         int flags = FMODE_WRITE;
1346         ENTRY;
1347
1348         /* Bug 1152: copy properly when this is no longer true */
1349         LASSERT(sizeof(lum) == sizeof(*lump));
1350         LASSERT(sizeof(lum.lmm_objects[0]) == sizeof(lump->lmm_objects[0]));
1351         rc = copy_from_user(&lum, lump, sizeof(lum));
1352         if (rc)
1353                 RETURN(-EFAULT);
1354
1355         rc = ll_lov_setstripe_ea_info(inode, file, flags, &lum, sizeof(lum));
1356         if (rc == 0) {
1357                  put_user(0, &lump->lmm_stripe_count);
1358                  rc = obd_iocontrol(LL_IOC_LOV_GETSTRIPE, ll_i2dtexp(inode),
1359                                     0, ll_i2info(inode)->lli_smd, lump);
1360         }
1361         RETURN(rc);
1362 }
1363
1364 static int ll_lov_getstripe(struct inode *inode, unsigned long arg)
1365 {
1366         struct lov_stripe_md *lsm = ll_i2info(inode)->lli_smd;
1367
1368         if (!lsm)
1369                 RETURN(-ENODATA);
1370
1371         return obd_iocontrol(LL_IOC_LOV_GETSTRIPE, ll_i2dtexp(inode), 0, lsm,
1372                              (void *)arg);
1373 }
1374
1375 static int ll_get_grouplock(struct inode *inode, struct file *file,
1376                          unsigned long arg)
1377 {
1378         struct ll_file_data *fd = file->private_data;
1379         ldlm_policy_data_t policy = { .l_extent = { .start = 0,
1380                                                     .end = OBD_OBJECT_EOF}};
1381         struct lustre_handle lockh = { 0 };
1382         struct ll_inode_info *lli = ll_i2info(inode);
1383         struct lov_stripe_md *lsm = lli->lli_smd;
1384         int flags = 0, rc;
1385         ENTRY;
1386
1387         if (fd->fd_flags & LL_FILE_GROUP_LOCKED) {
1388                 RETURN(-EINVAL);
1389         }
1390
1391         policy.l_extent.gid = arg;
1392         if (file->f_flags & O_NONBLOCK)
1393                 flags = LDLM_FL_BLOCK_NOWAIT;
1394
1395         rc = ll_extent_lock(fd, inode, lsm, LCK_GROUP, &policy, &lockh, flags,
1396                             &ll_i2sbi(inode)->ll_grouplock_stime);
1397         if (rc != 0)
1398                 RETURN(rc);
1399
1400         fd->fd_flags |= LL_FILE_GROUP_LOCKED|LL_FILE_IGNORE_LOCK;
1401         fd->fd_gid = arg;
1402         memcpy(&fd->fd_cwlockh, &lockh, sizeof(lockh));
1403
1404         RETURN(0);
1405 }
1406
1407 static int ll_put_grouplock(struct inode *inode, struct file *file,
1408                          unsigned long arg)
1409 {
1410         struct ll_file_data *fd = file->private_data;
1411         struct ll_inode_info *lli = ll_i2info(inode);
1412         struct lov_stripe_md *lsm = lli->lli_smd;
1413         int rc;
1414         ENTRY;
1415
1416         if (!(fd->fd_flags & LL_FILE_GROUP_LOCKED)) {
1417                 /* Ugh, it's already unlocked. */
1418                 RETURN(-EINVAL);
1419         }
1420
1421         if (fd->fd_gid != arg) /* Ugh? Unlocking with different gid? */
1422                 RETURN(-EINVAL);
1423
1424         fd->fd_flags &= ~(LL_FILE_GROUP_LOCKED|LL_FILE_IGNORE_LOCK);
1425
1426         rc = ll_extent_unlock(fd, inode, lsm, LCK_GROUP, &fd->fd_cwlockh);
1427         if (rc)
1428                 RETURN(rc);
1429
1430         fd->fd_gid = 0;
1431         memset(&fd->fd_cwlockh, 0, sizeof(fd->fd_cwlockh));
1432
1433         RETURN(0);
1434 }
1435
1436 int ll_file_ioctl(struct inode *inode, struct file *file, unsigned int cmd,
1437                   unsigned long arg)
1438 {
1439         struct ll_file_data *fd = file->private_data;
1440         struct ll_sb_info *sbi = ll_i2sbi(inode);
1441         int flags;
1442         ENTRY;
1443
1444         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p),cmd=%x\n", inode->i_ino,
1445                inode->i_generation, inode, cmd);
1446
1447         if (_IOC_TYPE(cmd) == 'T') /* tty ioctls */
1448                 RETURN(-ENOTTY);
1449
1450         lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats, LPROC_LL_IOCTL);
1451         switch(cmd) {
1452         case LL_IOC_GETFLAGS:
1453                 /* Get the current value of the file flags */
1454                 return put_user(fd->fd_flags, (int *)arg);
1455         case LL_IOC_SETFLAGS:
1456         case LL_IOC_CLRFLAGS:
1457                 /* Set or clear specific file flags */
1458                 /* XXX This probably needs checks to ensure the flags are
1459                  *     not abused, and to handle any flag side effects.
1460                  */
1461                 if (get_user(flags, (int *) arg))
1462                         RETURN(-EFAULT);
1463
1464                 if (cmd == LL_IOC_SETFLAGS)
1465                         fd->fd_flags |= flags;
1466                 else
1467                         fd->fd_flags &= ~flags;
1468                 RETURN(0);
1469         case LL_IOC_LOV_SETSTRIPE:
1470                 RETURN(ll_lov_setstripe(inode, file, arg));
1471         case LL_IOC_LOV_SETEA:
1472                 RETURN(ll_lov_setea(inode, file, arg));
1473         case IOC_MDC_SHOWFID: {
1474                 struct lustre_id *idp = (struct lustre_id *)arg;
1475                 struct lustre_id id;
1476                 char *filename;
1477                 int rc;
1478
1479                 filename = getname((const char *)arg);
1480                 if (IS_ERR(filename))
1481                         RETURN(PTR_ERR(filename));
1482
1483                 ll_inode2id(&id, inode);
1484
1485                 rc = ll_get_fid(sbi->ll_md_exp, &id, filename, &id);
1486                 if (rc < 0)
1487                         GOTO(out_filename, rc);
1488
1489                 rc = copy_to_user(idp, &id, sizeof(*idp));
1490                 if (rc)
1491                         GOTO(out_filename, rc = -EFAULT);
1492
1493                 EXIT;
1494         out_filename:
1495                 putname(filename);
1496                 return rc;
1497         }
1498         case LL_IOC_LOV_GETSTRIPE:
1499                 RETURN(ll_lov_getstripe(inode, arg));
1500         case LL_IOC_RECREATE_OBJ:
1501                 RETURN(ll_lov_recreate_obj(inode, file, arg));
1502         case EXT3_IOC_GETFLAGS:
1503         case EXT3_IOC_SETFLAGS:
1504                 RETURN( ll_iocontrol(inode, file, cmd, arg) );
1505         case LL_IOC_GROUP_LOCK:
1506                 RETURN(ll_get_grouplock(inode, file, arg));
1507         case LL_IOC_GROUP_UNLOCK:
1508                 RETURN(ll_put_grouplock(inode, file, arg));
1509         case EXT3_IOC_GETVERSION_OLD:
1510         case EXT3_IOC_GETVERSION:
1511                 return put_user(inode->i_generation, (int *) arg);
1512         /* We need to special case any other ioctls we want to handle,
1513          * to send them to the MDS/OST as appropriate and to properly
1514          * network encode the arg field.
1515         case EXT2_IOC_GETVERSION_OLD:
1516         case EXT2_IOC_GETVERSION_NEW:
1517         case EXT2_IOC_SETVERSION_OLD:
1518         case EXT2_IOC_SETVERSION_NEW:
1519         case EXT3_IOC_SETVERSION_OLD:
1520         case EXT3_IOC_SETVERSION:
1521         */
1522         default:
1523                 RETURN( obd_iocontrol(cmd, ll_i2dtexp(inode), 0, NULL,
1524                                       (void *)arg) );
1525         }
1526 }
1527
1528 loff_t ll_file_seek(struct file *file, loff_t offset, int origin)
1529 {
1530         struct inode *inode = file->f_dentry->d_inode;
1531         struct ll_file_data *fd = file->private_data;
1532         struct lov_stripe_md *lsm = ll_i2info(inode)->lli_smd;
1533         struct lustre_handle lockh = {0};
1534         loff_t retval;
1535         ENTRY;
1536         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p),to=%llu\n", inode->i_ino,
1537                inode->i_generation, inode,
1538                offset + ((origin==2) ? inode->i_size : file->f_pos));
1539
1540         lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats, LPROC_LL_LLSEEK);
1541         if (origin == 2) { /* SEEK_END */
1542                 ldlm_policy_data_t policy = { .l_extent = {0, OBD_OBJECT_EOF }};
1543                 struct ll_inode_info *lli = ll_i2info(inode);
1544                 int nonblock = 0, rc;
1545
1546                 if (file->f_flags & O_NONBLOCK)
1547                         nonblock = LDLM_FL_BLOCK_NOWAIT;
1548
1549                 rc = ll_extent_lock(fd, inode, lsm, LCK_PR, &policy, &lockh,
1550                                     nonblock, &ll_i2sbi(inode)->ll_seek_stime);
1551                 if (rc != 0)
1552                         RETURN(rc);
1553
1554                 down(&lli->lli_size_sem);
1555                 offset += inode->i_size;
1556                 up(&lli->lli_size_sem);
1557         } else if (origin == 1) { /* SEEK_CUR */
1558                 offset += file->f_pos;
1559         }
1560
1561         retval = -EINVAL;
1562         if (offset >= 0 && offset <= ll_file_maxbytes(inode)) {
1563                 if (offset != file->f_pos) {
1564                         file->f_pos = offset;
1565 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
1566                         file->f_reada = 0;
1567                         file->f_version = ++event;
1568 #endif
1569                 }
1570                 retval = offset;
1571         }
1572
1573         if (origin == 2)
1574                 ll_extent_unlock(fd, inode, lsm, LCK_PR, &lockh);
1575         RETURN(retval);
1576 }
1577
1578 int ll_fsync(struct file *file, struct dentry *dentry, int data)
1579 {
1580         struct inode *inode = dentry->d_inode;
1581         struct lov_stripe_md *lsm = ll_i2info(inode)->lli_smd;
1582         struct lustre_id id;
1583         struct ptlrpc_request *req;
1584         int rc, err;
1585         ENTRY;
1586         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p)\n", inode->i_ino,
1587                inode->i_generation, inode);
1588
1589         lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats, LPROC_LL_FSYNC);
1590
1591         /* fsync's caller has already called _fdata{sync,write}, we want
1592          * that IO to finish before calling the osc and mdc sync methods */
1593         rc = filemap_fdatawait(inode->i_mapping);
1594
1595         ll_inode2id(&id, inode);
1596         err = md_sync(ll_i2sbi(inode)->ll_md_exp, &id, &req);
1597         if (!rc)
1598                 rc = err;
1599         if (!err)
1600                 ptlrpc_req_finished(req);
1601
1602         if (data && lsm) {
1603                 struct obdo *oa = obdo_alloc();
1604
1605                 if (!oa)
1606                         RETURN(rc ? rc : -ENOMEM);
1607
1608                 oa->o_id = lsm->lsm_object_id;
1609                 oa->o_gr = lsm->lsm_object_gr;
1610                 oa->o_valid = OBD_MD_FLID;
1611                 obdo_from_inode(oa, inode, (OBD_MD_FLTYPE | OBD_MD_FLATIME |
1612                                             OBD_MD_FLMTIME | OBD_MD_FLCTIME |
1613                                             OBD_MD_FLGROUP));
1614
1615                 err = obd_sync(ll_i2sbi(inode)->ll_dt_exp, oa, lsm,
1616                                0, OBD_OBJECT_EOF);
1617                 if (!rc)
1618                         rc = err;
1619                 obdo_free(oa);
1620         }
1621
1622         RETURN(rc);
1623 }
1624
1625 int ll_file_flock(struct file *file, int cmd, struct file_lock *file_lock)
1626 {
1627         struct inode *inode = file->f_dentry->d_inode;
1628         struct ll_inode_info *li = ll_i2info(inode);
1629         struct ll_sb_info *sbi = ll_i2sbi(inode);
1630         struct obd_device *obddev;
1631         struct ldlm_res_id res_id =
1632                 { .name = {id_fid(&li->lli_id), id_group(&li->lli_id), LDLM_FLOCK} };
1633         struct lustre_handle lockh = {0};
1634         ldlm_policy_data_t flock;
1635         ldlm_mode_t mode = 0;
1636         int flags = 0;
1637         int rc;
1638         ENTRY;
1639
1640         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu file_lock=%p\n",
1641                inode->i_ino, file_lock);
1642
1643         flock.l_flock.pid = file_lock->fl_pid;
1644         flock.l_flock.start = file_lock->fl_start;
1645         flock.l_flock.end = file_lock->fl_end;
1646
1647         switch (file_lock->fl_type) {
1648         case F_RDLCK:
1649                 mode = LCK_PR;
1650                 break;
1651         case F_UNLCK:
1652                 /* An unlock request may or may not have any relation to
1653                  * existing locks so we may not be able to pass a lock handle
1654                  * via a normal ldlm_lock_cancel() request. The request may even
1655                  * unlock a byte range in the middle of an existing lock. In
1656                  * order to process an unlock request we need all of the same
1657                  * information that is given with a normal read or write record
1658                  * lock request. To avoid creating another ldlm unlock (cancel)
1659                  * message we'll treat a LCK_NL flock request as an unlock. */
1660                 mode = LCK_NL;
1661                 break;
1662         case F_WRLCK:
1663                 mode = LCK_PW;
1664                 break;
1665         default:
1666                 CERROR("unknown fcntl lock type: %d\n", file_lock->fl_type);
1667                 LBUG();
1668         }
1669
1670         switch (cmd) {
1671         case F_SETLKW:
1672 #ifdef F_SETLKW64
1673         case F_SETLKW64:
1674 #endif
1675                 flags = 0;
1676                 break;
1677         case F_SETLK:
1678 #ifdef F_SETLK64
1679         case F_SETLK64:
1680 #endif
1681                 flags = LDLM_FL_BLOCK_NOWAIT;
1682                 break;
1683         case F_GETLK:
1684 #ifdef F_GETLK64
1685         case F_GETLK64:
1686 #endif
1687                 flags = LDLM_FL_TEST_LOCK;
1688                 /* Save the old mode so that if the mode in the lock changes we
1689                  * can decrement the appropriate reader or writer refcount. */
1690                 file_lock->fl_type = mode;
1691                 break;
1692         default:
1693                 CERROR("unknown fcntl lock command: %d\n", cmd);
1694                 LBUG();
1695         }
1696
1697         CDEBUG(D_DLMTRACE, "inode=%lu, pid="LPU64", flags=%#x, mode=%u, "
1698                "start="LPU64", end="LPU64"\n", inode->i_ino, flock.l_flock.pid,
1699                flags, mode, flock.l_flock.start, flock.l_flock.end);
1700
1701         obddev = md_get_real_obd(sbi->ll_md_exp, &li->lli_id);
1702         rc = ldlm_cli_enqueue(obddev->obd_self_export, NULL,
1703                               obddev->obd_namespace,
1704                               res_id, LDLM_FLOCK, &flock, mode, &flags,
1705                               NULL, ldlm_flock_completion_ast, NULL, file_lock,
1706                               NULL, 0, NULL, &lockh);
1707         RETURN(rc);
1708 }
1709
1710 int ll_inode_revalidate_it(struct dentry *dentry)
1711 {
1712         struct lookup_intent oit = { .it_op = IT_GETATTR };
1713         struct inode *inode = dentry->d_inode;
1714         struct ptlrpc_request *req = NULL;
1715         struct ll_inode_info *lli;
1716         struct lov_stripe_md *lsm;
1717         struct ll_sb_info *sbi;
1718         struct lustre_id id;
1719         int rc;
1720         ENTRY;
1721
1722         if (!inode) {
1723                 CERROR("REPORT THIS LINE TO PETER\n");
1724                 RETURN(0);
1725         }
1726
1727         sbi = ll_i2sbi(inode);
1728
1729         ll_inode2id(&id, inode);
1730         lli = ll_i2info(inode);
1731         LASSERT(id_fid(&id) != 0);
1732
1733         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p), name=%s(%p)\n",
1734                inode->i_ino, inode->i_generation, inode, dentry->d_name.name,
1735                dentry);
1736
1737 #if (LINUX_VERSION_CODE <= KERNEL_VERSION(2,5,0))
1738         lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats, LPROC_LL_REVALIDATE);
1739 #endif
1740
1741         rc = ll_intent_alloc(&oit);
1742         if (rc)
1743                 RETURN(-ENOMEM);
1744
1745         rc = md_intent_lock(sbi->ll_md_exp, &id, NULL, 0, NULL, 0, &id,
1746                             &oit, 0, &req, ll_mdc_blocking_ast);
1747         if (rc < 0)
1748                 GOTO(out, rc);
1749
1750         rc = revalidate_it_finish(req, 1, &oit, dentry);
1751         if (rc) {
1752                 GOTO(out, rc);
1753         }
1754
1755         ll_lookup_finish_locks(&oit, dentry);
1756
1757         lsm = lli->lli_smd;
1758         if (lsm == NULL) /* object not yet allocated, don't validate size */
1759                 GOTO(out, rc = 0);
1760
1761         /*
1762          * ll_glimpse_size() will prefer locally cached writes if they extend
1763          * the file.
1764          */
1765         rc = ll_glimpse_size(inode);
1766         EXIT;
1767 out:
1768         ll_intent_release(&oit);
1769         if (req)
1770                 ptlrpc_req_finished(req);
1771         return rc;
1772 }
1773
1774 #if (LINUX_VERSION_CODE > KERNEL_VERSION(2,5,0))
1775 int ll_getattr(struct vfsmount *mnt, struct dentry *de, struct kstat *stat)
1776 {
1777         int res = 0;
1778         struct inode *inode = de->d_inode;
1779         struct ll_inode_info *lli = ll_i2info(inode);
1780
1781         res = ll_inode_revalidate_it(de);
1782         lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats, LPROC_LL_GETATTR);
1783
1784         if (res)
1785                 return res;
1786
1787         stat->ino = inode->i_ino;
1788         stat->mode = inode->i_mode;
1789         stat->nlink = inode->i_nlink;
1790         stat->uid = inode->i_uid;
1791         stat->gid = inode->i_gid;
1792         stat->atime = inode->i_atime;
1793         stat->mtime = inode->i_mtime;
1794         stat->ctime = inode->i_ctime;
1795         stat->blksize = inode->i_blksize;
1796
1797         down(&lli->lli_size_sem);
1798         stat->size = inode->i_size;
1799         stat->blocks = inode->i_blocks;
1800         up(&lli->lli_size_sem);
1801
1802         stat->rdev = kdev_t_to_nr(inode->i_rdev);
1803         stat->dev = id_group(&ll_i2info(inode)->lli_id);
1804         return 0;
1805 }
1806 #endif
1807
1808 static
1809 int ll_setxattr_internal(struct inode *inode, const char *name,
1810                          const void *value, size_t size, int flags,
1811                          __u64 valid)
1812 {
1813         struct ll_sb_info *sbi = ll_i2sbi(inode);
1814         struct ptlrpc_request *request = NULL;
1815         struct mdc_op_data op_data;
1816         struct iattr attr;
1817         int rc = 0;
1818         ENTRY;
1819
1820         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu\n", inode->i_ino);
1821         lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats, LPROC_LL_SETXATTR);
1822
1823         memset(&attr, 0x0, sizeof(attr));
1824         attr.ia_valid |= valid;
1825         attr.ia_attr_flags = flags;
1826
1827         ll_prepare_mdc_data(&op_data, inode, NULL, NULL, 0, 0);
1828
1829         rc = md_setattr(sbi->ll_md_exp, &op_data, &attr,
1830                         (void*) name, strnlen(name, XATTR_NAME_MAX)+1,
1831                         (void*) value, size, &request);
1832         if (rc) {
1833                 CERROR("md_setattr fails: rc = %d\n", rc);
1834                 GOTO(out, rc);
1835         }
1836
1837  out:
1838         ptlrpc_req_finished(request);
1839         RETURN(rc);
1840 }
1841
1842 int ll_setxattr(struct dentry *dentry, const char *name, const void *value,
1843                 size_t size, int flags)
1844 {
1845         int rc, error;
1846         struct posix_acl *acl;
1847         struct ll_inode_info *lli;
1848         ENTRY;
1849
1850         rc = ll_setxattr_internal(dentry->d_inode, name, value, size,
1851                                   flags, ATTR_EA);
1852
1853         /* update inode's acl info */
1854         if (rc == 0 && strcmp(name, XATTR_NAME_ACL_ACCESS) == 0) {
1855                 if (value) {
1856                         acl = posix_acl_from_xattr(value, size);
1857                         if (IS_ERR(acl)) {
1858                                 CERROR("convert from xattr to acl error: %ld",
1859                                         PTR_ERR(acl));
1860                                 GOTO(out, rc);
1861                         } else if (acl) {
1862                                 error = posix_acl_valid(acl);
1863                                 if (error) {
1864                                         CERROR("acl valid error: %d", error);
1865                                         posix_acl_release(acl);
1866                                         GOTO(out, rc);
1867                                 }
1868                         }
1869                 } else {
1870                         acl = NULL;
1871                 }
1872
1873                 lli = ll_i2info(dentry->d_inode);
1874                 spin_lock(&lli->lli_lock);
1875                 if (lli->lli_acl_access != NULL)
1876                         posix_acl_release(lli->lli_acl_access);
1877                 lli->lli_acl_access = acl;
1878                 spin_unlock(&lli->lli_lock);
1879         }
1880         EXIT;
1881 out:
1882         return(rc);
1883 }
1884
1885 int ll_removexattr(struct dentry *dentry, const char *name)
1886 {
1887         return ll_setxattr_internal(dentry->d_inode, name, NULL, 0, 0,
1888                                     ATTR_EA_RM);
1889 }
1890
1891 static
1892 int ll_getxattr_internal(struct inode *inode, const char *name, int namelen,
1893                          void *value, size_t size, __u64 valid)
1894 {
1895         struct ptlrpc_request *request = NULL;
1896         struct ll_sb_info *sbi = ll_i2sbi(inode);
1897         struct lustre_id id;
1898         struct mds_body *body;
1899         void *ea_data;
1900         int rc, ea_size;
1901         ENTRY;
1902
1903         lprocfs_counter_incr(ll_i2sbi(inode)->ll_stats, LPROC_LL_GETXATTR);
1904
1905         ll_inode2id(&id, inode);
1906         rc = md_getattr(sbi->ll_md_exp, &id, valid, name, namelen,
1907                          size, &request);
1908         if (rc) {
1909                 if (rc != -ENODATA && rc != -EOPNOTSUPP)
1910                         CERROR("md_getattr fails: rc = %d\n", rc);
1911                 GOTO(out, rc);
1912         }
1913
1914         body = lustre_msg_buf(request->rq_repmsg, 0, sizeof(*body));
1915         LASSERT(body != NULL);
1916         LASSERT_REPSWABBED(request, 0);
1917
1918         ea_size = body->eadatasize;
1919         LASSERT(ea_size <= request->rq_repmsg->buflens[0]);
1920
1921         if (size == 0)
1922                 GOTO(out, rc = ea_size);
1923
1924         ea_data = lustre_msg_buf(request->rq_repmsg, 1, ea_size);
1925         LASSERT(ea_data != NULL);
1926         LASSERT_REPSWABBED(request, 1);
1927
1928         if (value)
1929                 memcpy(value, ea_data, ea_size);
1930         rc = ea_size;
1931  out:
1932         ptlrpc_req_finished(request);
1933         RETURN(rc);
1934 }
1935
1936 int ll_getxattr(struct dentry *dentry, const char *name, void *value,
1937                 size_t size)
1938 {
1939         return ll_getxattr_internal(dentry->d_inode, name, strlen(name) + 1,
1940                                     value, size, OBD_MD_FLEA);
1941 }
1942
1943 int ll_listxattr(struct dentry *dentry, char *list, size_t size)
1944 {
1945         return ll_getxattr_internal(dentry->d_inode, NULL, 0, list, size,
1946                                     OBD_MD_FLEALIST);
1947 }
1948
1949 /*
1950  * XXX We could choose not to check DLM lock. Leave the decision
1951  * to remote acl handling.
1952  */
1953 static int
1954 lustre_check_acl(struct inode *inode, int mask)
1955 {
1956         struct lookup_intent it = { .it_op = IT_GETATTR };
1957         struct dentry de = { .d_inode = inode };
1958         struct ll_sb_info *sbi;
1959         struct lustre_id id;
1960         struct ptlrpc_request *req = NULL;
1961         struct ll_inode_info *lli = ll_i2info(inode);
1962         struct posix_acl *acl;
1963         int rc;
1964         ENTRY;
1965
1966         sbi = ll_i2sbi(inode);
1967         ll_inode2id(&id, inode);
1968
1969         if (ll_intent_alloc(&it))
1970                 return -EACCES;
1971
1972         rc = md_intent_lock(sbi->ll_md_exp, &id, NULL, 0, NULL, 0, &id,
1973                             &it, 0, &req, ll_mdc_blocking_ast);
1974         if (rc < 0) {
1975                 ll_intent_free(&it);
1976                 GOTO(out, rc);
1977         }
1978
1979         rc = revalidate_it_finish(req, 1, &it, &de);
1980         if (rc) {
1981                 ll_intent_release(&it);
1982                 GOTO(out, rc);
1983         }
1984
1985         ll_lookup_finish_locks(&it, &de);
1986         ll_intent_free(&it);
1987
1988         spin_lock(&lli->lli_lock);
1989         acl = posix_acl_dup(ll_i2info(inode)->lli_acl_access);
1990         spin_unlock(&lli->lli_lock);
1991
1992         if (!acl)
1993                 GOTO(out, rc = -EAGAIN);
1994
1995         rc = posix_acl_permission(inode, acl, mask);
1996         posix_acl_release(acl);
1997
1998 out:
1999         if (req)
2000                 ptlrpc_req_finished(req);
2001
2002         RETURN(rc);
2003 }
2004
2005 int ll_inode_permission(struct inode *inode, int mask, struct nameidata *nd)
2006 {
2007         return generic_permission(inode, mask, lustre_check_acl);
2008 }
2009
2010 struct file_operations ll_file_operations = {
2011         .read           = ll_file_read,
2012         .write          = ll_file_write,
2013         .ioctl          = ll_file_ioctl,
2014         .open           = ll_file_open,
2015         .release        = ll_file_release,
2016         .mmap           = ll_file_mmap,
2017         .llseek         = ll_file_seek,
2018 #if (LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,0))
2019         .sendfile       = generic_file_sendfile,
2020 #endif
2021         .fsync          = ll_fsync,
2022         .lock           = ll_file_flock
2023 };
2024
2025 struct inode_operations ll_file_inode_operations = {
2026         .setattr        = ll_setattr,
2027         .truncate       = ll_truncate,
2028 #if (LINUX_VERSION_CODE > KERNEL_VERSION(2,5,0))
2029         .getattr        = ll_getattr,
2030 #else
2031         .revalidate_it  = ll_inode_revalidate_it,
2032 #endif
2033         .setxattr       = ll_setxattr,
2034         .getxattr       = ll_getxattr,
2035         .listxattr      = ll_listxattr,
2036         .removexattr    = ll_removexattr,
2037         .permission     = ll_inode_permission,
2038 };
2039