lustre/llite/file.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2013, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/llite/file.c
  37  *
  38  * Author: Peter Braam <braam@clusterfs.com>
  39  * Author: Phil Schwan <phil@clusterfs.com>
  40  * Author: Andreas Dilger <adilger@clusterfs.com>
  41  */
  42
  43 #define DEBUG_SUBSYSTEM S_LLITE
  44 #include <lustre_dlm.h>
  45 #include <lustre_lite.h>
  46 #include <linux/pagemap.h>
  47 #include <linux/file.h>
  48 #include "llite_internal.h"
  49 #include <lustre/ll_fiemap.h>
  50
  51 #include "cl_object.h"
  52
  53 struct ll_file_data *ll_file_data_get(void)
  54 {
  55         struct ll_file_data *fd;
  56
  57         OBD_SLAB_ALLOC_PTR_GFP(fd, ll_file_data_slab, __GFP_IO);
  58         if (fd == NULL)
  59                 return NULL;
  60
  61         fd->fd_write_failed = false;
  62
  63         return fd;
  64 }
  65
  66 static void ll_file_data_put(struct ll_file_data *fd)
  67 {
  68         if (fd != NULL)
  69                 OBD_SLAB_FREE_PTR(fd, ll_file_data_slab);
  70 }
  71
  72 void ll_pack_inode2opdata(struct inode *inode, struct md_op_data *op_data,
  73                           struct lustre_handle *fh)
  74 {
  75         op_data->op_fid1 = ll_i2info(inode)->lli_fid;
  76         op_data->op_attr.ia_mode = inode->i_mode;
  77         op_data->op_attr.ia_atime = inode->i_atime;
  78         op_data->op_attr.ia_mtime = inode->i_mtime;
  79         op_data->op_attr.ia_ctime = inode->i_ctime;
  80         op_data->op_attr.ia_size = i_size_read(inode);
  81         op_data->op_attr_blocks = inode->i_blocks;
  82         ((struct ll_iattr *)&op_data->op_attr)->ia_attr_flags =
  83                                         ll_inode_to_ext_flags(inode->i_flags);
  84         op_data->op_ioepoch = ll_i2info(inode)->lli_ioepoch;
  85         if (fh)
  86                 op_data->op_handle = *fh;
  87         op_data->op_capa1 = ll_mdscapa_get(inode);
  88
  89         if (LLIF_DATA_MODIFIED & ll_i2info(inode)->lli_flags)
  90                 op_data->op_bias |= MDS_DATA_MODIFIED;
  91 }
  92
  93 /**
  94  * Closes the IO epoch and packs all the attributes into @op_data for
  95  * the CLOSE rpc.
  96  */
  97 static void ll_prepare_close(struct inode *inode, struct md_op_data *op_data,
  98                              struct obd_client_handle *och)
  99 {
 100         ENTRY;
 101
 102         op_data->op_attr.ia_valid = ATTR_MODE | ATTR_ATIME | ATTR_ATIME_SET |
 103                                         ATTR_MTIME | ATTR_MTIME_SET |
 104                                         ATTR_CTIME | ATTR_CTIME_SET;
 105
 106         if (!(och->och_flags & FMODE_WRITE))
 107                 goto out;
 108
 109         if (!exp_connect_som(ll_i2mdexp(inode)) || !S_ISREG(inode->i_mode))
 110                 op_data->op_attr.ia_valid |= ATTR_SIZE | ATTR_BLOCKS;
 111         else
 112                 ll_ioepoch_close(inode, op_data, &och, 0);
 113
 114 out:
 115         ll_pack_inode2opdata(inode, op_data, &och->och_fh);
 116         ll_prep_md_op_data(op_data, inode, NULL, NULL,
 117                            0, 0, LUSTRE_OPC_ANY, NULL);
 118         EXIT;
 119 }
 120
 121 static int ll_close_inode_openhandle(struct obd_export *md_exp,
 122                                      struct inode *inode,
 123                                      struct obd_client_handle *och)
 124 {
 125         struct obd_export *exp = ll_i2mdexp(inode);
 126         struct md_op_data *op_data;
 127         struct ptlrpc_request *req = NULL;
 128         struct obd_device *obd = class_exp2obd(exp);
 129         int epoch_close = 1;
 130         int rc;
 131         ENTRY;
 132
 133         if (obd == NULL) {
 134                 /*
 135                  * XXX: in case of LMV, is this correct to access
 136                  * ->exp_handle?
 137                  */
 138                 CERROR("Invalid MDC connection handle "LPX64"\n",
 139                        ll_i2mdexp(inode)->exp_handle.h_cookie);
 140                 GOTO(out, rc = 0);
 141         }
 142
 143         OBD_ALLOC_PTR(op_data);
 144         if (op_data == NULL)
 145                 GOTO(out, rc = -ENOMEM); // XXX We leak openhandle and request here.
 146
 147         ll_prepare_close(inode, op_data, och);
 148         epoch_close = (op_data->op_flags & MF_EPOCH_CLOSE);
 149         rc = md_close(md_exp, op_data, och->och_mod, &req);
 150         if (rc == -EAGAIN) {
 151                 /* This close must have the epoch closed. */
 152                 LASSERT(epoch_close);
 153                 /* MDS has instructed us to obtain Size-on-MDS attribute from
 154                  * OSTs and send setattr to back to MDS. */
 155                 rc = ll_som_update(inode, op_data);
 156                 if (rc) {
 157                         CERROR("inode %lu mdc Size-on-MDS update failed: "
 158                                "rc = %d\n", inode->i_ino, rc);
 159                         rc = 0;
 160                 }
 161         } else if (rc) {
 162                 CERROR("inode %lu mdc close failed: rc = %d\n",
 163                        inode->i_ino, rc);
 164         }
 165
 166         /* DATA_MODIFIED flag was successfully sent on close, cancel data
 167          * modification flag. */
 168         if (rc == 0 && (op_data->op_bias & MDS_DATA_MODIFIED)) {
 169                 struct ll_inode_info *lli = ll_i2info(inode);
 170
 171                 spin_lock(&lli->lli_lock);
 172                 lli->lli_flags &= ~LLIF_DATA_MODIFIED;
 173                 spin_unlock(&lli->lli_lock);
 174         }
 175
 176         ll_finish_md_op_data(op_data);
 177
 178         if (rc == 0) {
 179                 rc = ll_objects_destroy(req, inode);
 180                 if (rc)
 181                         CERROR("inode %lu ll_objects destroy: rc = %d\n",
 182                                inode->i_ino, rc);
 183         }
 184
 185         EXIT;
 186 out:
 187
 188         if (exp_connect_som(exp) && !epoch_close &&
 189             S_ISREG(inode->i_mode) && (och->och_flags & FMODE_WRITE)) {
 190                 ll_queue_done_writing(inode, LLIF_DONE_WRITING);
 191         } else {
 192                 md_clear_open_replay_data(md_exp, och);
 193                 /* Free @och if it is not waiting for DONE_WRITING. */
 194                 och->och_fh.cookie = DEAD_HANDLE_MAGIC;
 195                 OBD_FREE_PTR(och);
 196         }
 197         if (req) /* This is close request */
 198                 ptlrpc_req_finished(req);
 199         return rc;
 200 }
 201
 202 int ll_md_real_close(struct inode *inode, int flags)
 203 {
 204         struct ll_inode_info *lli = ll_i2info(inode);
 205         struct obd_client_handle **och_p;
 206         struct obd_client_handle *och;
 207         __u64 *och_usecount;
 208         int rc = 0;
 209         ENTRY;
 210
 211         if (flags & FMODE_WRITE) {
 212                 och_p = &lli->lli_mds_write_och;
 213                 och_usecount = &lli->lli_open_fd_write_count;
 214         } else if (flags & FMODE_EXEC) {
 215                 och_p = &lli->lli_mds_exec_och;
 216                 och_usecount = &lli->lli_open_fd_exec_count;
 217         } else {
 218                 LASSERT(flags & FMODE_READ);
 219                 och_p = &lli->lli_mds_read_och;
 220                 och_usecount = &lli->lli_open_fd_read_count;
 221         }
 222
 223         mutex_lock(&lli->lli_och_mutex);
 224         if (*och_usecount) { /* There are still users of this handle, so
 225                                 skip freeing it. */
 226                 mutex_unlock(&lli->lli_och_mutex);
 227                 RETURN(0);
 228         }
 229         och=*och_p;
 230         *och_p = NULL;
 231         mutex_unlock(&lli->lli_och_mutex);
 232
 233         if (och) { /* There might be a race and somebody have freed this och
 234                       already */
 235                 rc = ll_close_inode_openhandle(ll_i2sbi(inode)->ll_md_exp,
 236                                                inode, och);
 237         }
 238
 239         RETURN(rc);
 240 }
 241
 242 int ll_md_close(struct obd_export *md_exp, struct inode *inode,
 243                 struct file *file)
 244 {
 245         struct ll_file_data *fd = LUSTRE_FPRIVATE(file);
 246         struct ll_inode_info *lli = ll_i2info(inode);
 247         int rc = 0;
 248         ENTRY;
 249
 250         /* clear group lock, if present */
 251         if (unlikely(fd->fd_flags & LL_FILE_GROUP_LOCKED))
 252                 ll_put_grouplock(inode, file, fd->fd_grouplock.cg_gid);
 253
 254         if (fd->fd_lease_och != NULL) {
 255                 bool lease_broken;
 256
 257                 /* Usually the lease is not released when the
 258                  * application crashed, we need to release here. */
 259                 rc = ll_lease_close(fd->fd_lease_och, inode, &lease_broken);
 260                 CDEBUG(rc ? D_ERROR : D_INODE, "Clean up lease "DFID" %d/%d\n",
 261                         PFID(&lli->lli_fid), rc, lease_broken);
 262
 263                 fd->fd_lease_och = NULL;
 264         }
 265
 266         if (fd->fd_och != NULL) {
 267                 rc = ll_close_inode_openhandle(md_exp, inode, fd->fd_och);
 268                 fd->fd_och = NULL;
 269                 GOTO(out, rc);
 270         }
 271
 272         /* Let's see if we have good enough OPEN lock on the file and if
 273            we can skip talking to MDS */
 274         if (file->f_dentry->d_inode) { /* Can this ever be false? */
 275                 int lockmode;
 276                 int flags = LDLM_FL_BLOCK_GRANTED | LDLM_FL_TEST_LOCK;
 277                 struct lustre_handle lockh;
 278                 struct inode *inode = file->f_dentry->d_inode;
 279                 ldlm_policy_data_t policy = {.l_inodebits={MDS_INODELOCK_OPEN}};
 280
 281                 mutex_lock(&lli->lli_och_mutex);
 282                 if (fd->fd_omode & FMODE_WRITE) {
 283                         lockmode = LCK_CW;
 284                         LASSERT(lli->lli_open_fd_write_count);
 285                         lli->lli_open_fd_write_count--;
 286                 } else if (fd->fd_omode & FMODE_EXEC) {
 287                         lockmode = LCK_PR;
 288                         LASSERT(lli->lli_open_fd_exec_count);
 289                         lli->lli_open_fd_exec_count--;
 290                 } else {
 291                         lockmode = LCK_CR;
 292                         LASSERT(lli->lli_open_fd_read_count);
 293                         lli->lli_open_fd_read_count--;
 294                 }
 295                 mutex_unlock(&lli->lli_och_mutex);
 296
 297                 if (!md_lock_match(md_exp, flags, ll_inode2fid(inode),
 298                                    LDLM_IBITS, &policy, lockmode,
 299                                    &lockh)) {
 300                         rc = ll_md_real_close(file->f_dentry->d_inode,
 301                                               fd->fd_omode);
 302                 }
 303         } else {
 304                 CERROR("Releasing a file %p with negative dentry %p. Name %s",
 305                        file, file->f_dentry, file->f_dentry->d_name.name);
 306         }
 307
 308 out:
 309         LUSTRE_FPRIVATE(file) = NULL;
 310         ll_file_data_put(fd);
 311         ll_capa_close(inode);
 312
 313         RETURN(rc);
 314 }
 315
 316 /* While this returns an error code, fput() the caller does not, so we need
 317  * to make every effort to clean up all of our state here.  Also, applications
 318  * rarely check close errors and even if an error is returned they will not
 319  * re-try the close call.
 320  */
 321 int ll_file_release(struct inode *inode, struct file *file)
 322 {
 323         struct ll_file_data *fd;
 324         struct ll_sb_info *sbi = ll_i2sbi(inode);
 325         struct ll_inode_info *lli = ll_i2info(inode);
 326         int rc;
 327         ENTRY;
 328
 329         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p)\n", inode->i_ino,
 330                inode->i_generation, inode);
 331
 332 #ifdef CONFIG_FS_POSIX_ACL
 333         if (sbi->ll_flags & LL_SBI_RMT_CLIENT &&
 334             inode == inode->i_sb->s_root->d_inode) {
 335                 struct ll_file_data *fd = LUSTRE_FPRIVATE(file);
 336
 337                 LASSERT(fd != NULL);
 338                 if (unlikely(fd->fd_flags & LL_FILE_RMTACL)) {
 339                         fd->fd_flags &= ~LL_FILE_RMTACL;
 340                         rct_del(&sbi->ll_rct, cfs_curproc_pid());
 341                         et_search_free(&sbi->ll_et, cfs_curproc_pid());
 342                 }
 343         }
 344 #endif
 345
 346         if (inode->i_sb->s_root != file->f_dentry)
 347                 ll_stats_ops_tally(sbi, LPROC_LL_RELEASE, 1);
 348         fd = LUSTRE_FPRIVATE(file);
 349         LASSERT(fd != NULL);
 350
 351         /* The last ref on @file, maybe not the the owner pid of statahead.
 352          * Different processes can open the same dir, "ll_opendir_key" means:
 353          * it is me that should stop the statahead thread. */
 354         if (S_ISDIR(inode->i_mode) && lli->lli_opendir_key == fd &&
 355             lli->lli_opendir_pid != 0)
 356                 ll_stop_statahead(inode, lli->lli_opendir_key);
 357
 358         if (inode->i_sb->s_root == file->f_dentry) {
 359                 LUSTRE_FPRIVATE(file) = NULL;
 360                 ll_file_data_put(fd);
 361                 RETURN(0);
 362         }
 363
 364         if (!S_ISDIR(inode->i_mode)) {
 365                 lov_read_and_clear_async_rc(lli->lli_clob);
 366                 lli->lli_async_rc = 0;
 367         }
 368
 369         rc = ll_md_close(sbi->ll_md_exp, inode, file);
 370
 371         if (CFS_FAIL_TIMEOUT_MS(OBD_FAIL_PTLRPC_DUMP_LOG, cfs_fail_val))
 372                 libcfs_debug_dumplog();
 373
 374         RETURN(rc);
 375 }
 376
 377 static int ll_intent_file_open(struct file *file, void *lmm,
 378                                int lmmsize, struct lookup_intent *itp)
 379 {
 380         struct ll_sb_info *sbi = ll_i2sbi(file->f_dentry->d_inode);
 381         struct dentry *parent = file->f_dentry->d_parent;
 382         const char *name = file->f_dentry->d_name.name;
 383         const int len = file->f_dentry->d_name.len;
 384         struct md_op_data *op_data;
 385         struct ptlrpc_request *req;
 386         __u32 opc = LUSTRE_OPC_ANY;
 387         int rc;
 388         ENTRY;
 389
 390         if (!parent)
 391                 RETURN(-ENOENT);
 392
 393         /* Usually we come here only for NFSD, and we want open lock.
 394            But we can also get here with pre 2.6.15 patchless kernels, and in
 395            that case that lock is also ok */
 396         /* We can also get here if there was cached open handle in revalidate_it
 397          * but it disappeared while we were getting from there to ll_file_open.
 398          * But this means this file was closed and immediatelly opened which
 399          * makes a good candidate for using OPEN lock */
 400         /* If lmmsize & lmm are not 0, we are just setting stripe info
 401          * parameters. No need for the open lock */
 402         if (lmm == NULL && lmmsize == 0) {
 403                 itp->it_flags |= MDS_OPEN_LOCK;
 404                 if (itp->it_flags & FMODE_WRITE)
 405                         opc = LUSTRE_OPC_CREATE;
 406         }
 407
 408         op_data  = ll_prep_md_op_data(NULL, parent->d_inode,
 409                                       file->f_dentry->d_inode, name, len,
 410                                       O_RDWR, opc, NULL);
 411         if (IS_ERR(op_data))
 412                 RETURN(PTR_ERR(op_data));
 413
 414         itp->it_flags |= MDS_OPEN_BY_FID;
 415         rc = md_intent_lock(sbi->ll_md_exp, op_data, lmm, lmmsize, itp,
 416                             0 /*unused */, &req, ll_md_blocking_ast, 0);
 417         ll_finish_md_op_data(op_data);
 418         if (rc == -ESTALE) {
 419                 /* reason for keep own exit path - don`t flood log
 420                 * with messages with -ESTALE errors.
 421                 */
 422                 if (!it_disposition(itp, DISP_OPEN_OPEN) ||
 423                      it_open_error(DISP_OPEN_OPEN, itp))
 424                         GOTO(out, rc);
 425                 ll_release_openhandle(file->f_dentry, itp);
 426                 GOTO(out, rc);
 427         }
 428
 429         if (it_disposition(itp, DISP_LOOKUP_NEG))
 430                 GOTO(out, rc = -ENOENT);
 431
 432         if (rc != 0 || it_open_error(DISP_OPEN_OPEN, itp)) {
 433                 rc = rc ? rc : it_open_error(DISP_OPEN_OPEN, itp);
 434                 CDEBUG(D_VFSTRACE, "lock enqueue: err: %d\n", rc);
 435                 GOTO(out, rc);
 436         }
 437
 438         rc = ll_prep_inode(&file->f_dentry->d_inode, req, NULL, itp);
 439         if (!rc && itp->d.lustre.it_lock_mode)
 440                 ll_set_lock_data(sbi->ll_md_exp, file->f_dentry->d_inode,
 441                                  itp, NULL);
 442
 443 out:
 444         ptlrpc_req_finished(itp->d.lustre.it_data);
 445         it_clear_disposition(itp, DISP_ENQ_COMPLETE);
 446         ll_intent_drop_lock(itp);
 447
 448         RETURN(rc);
 449 }
 450
 451 /**
 452  * Assign an obtained @ioepoch to client's inode. No lock is needed, MDS does
 453  * not believe attributes if a few ioepoch holders exist. Attributes for
 454  * previous ioepoch if new one is opened are also skipped by MDS.
 455  */
 456 void ll_ioepoch_open(struct ll_inode_info *lli, __u64 ioepoch)
 457 {
 458         if (ioepoch && lli->lli_ioepoch != ioepoch) {
 459                 lli->lli_ioepoch = ioepoch;
 460                 CDEBUG(D_INODE, "Epoch "LPU64" opened on "DFID"\n",
 461                        ioepoch, PFID(&lli->lli_fid));
 462         }
 463 }
 464
 465 static int ll_och_fill(struct obd_export *md_exp, struct lookup_intent *it,
 466                        struct obd_client_handle *och)
 467 {
 468         struct ptlrpc_request *req = it->d.lustre.it_data;
 469         struct mdt_body *body;
 470
 471         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 472         och->och_fh = body->handle;
 473         och->och_fid = body->fid1;
 474         och->och_lease_handle.cookie = it->d.lustre.it_lock_handle;
 475         och->och_magic = OBD_CLIENT_HANDLE_MAGIC;
 476         och->och_flags = it->it_flags;
 477
 478         return md_set_open_replay_data(md_exp, och, req);
 479 }
 480
 481 int ll_local_open(struct file *file, struct lookup_intent *it,
 482                   struct ll_file_data *fd, struct obd_client_handle *och)
 483 {
 484         struct inode *inode = file->f_dentry->d_inode;
 485         struct ll_inode_info *lli = ll_i2info(inode);
 486         ENTRY;
 487
 488         LASSERT(!LUSTRE_FPRIVATE(file));
 489
 490         LASSERT(fd != NULL);
 491
 492         if (och) {
 493                 struct ptlrpc_request *req = it->d.lustre.it_data;
 494                 struct mdt_body *body;
 495                 int rc;
 496
 497                 rc = ll_och_fill(ll_i2sbi(inode)->ll_md_exp, it, och);
 498                 if (rc != 0)
 499                         RETURN(rc);
 500
 501                 body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 502                 ll_ioepoch_open(lli, body->ioepoch);
 503         }
 504
 505         LUSTRE_FPRIVATE(file) = fd;
 506         ll_readahead_init(inode, &fd->fd_ras);
 507         fd->fd_omode = it->it_flags & (FMODE_READ | FMODE_WRITE | FMODE_EXEC);
 508
 509         RETURN(0);
 510 }
 511
 512 /* Open a file, and (for the very first open) create objects on the OSTs at
 513  * this time.  If opened with O_LOV_DELAY_CREATE, then we don't do the object
 514  * creation or open until ll_lov_setstripe() ioctl is called.
 515  *
 516  * If we already have the stripe MD locally then we don't request it in
 517  * md_open(), by passing a lmm_size = 0.
 518  *
 519  * It is up to the application to ensure no other processes open this file
 520  * in the O_LOV_DELAY_CREATE case, or the default striping pattern will be
 521  * used.  We might be able to avoid races of that sort by getting lli_open_sem
 522  * before returning in the O_LOV_DELAY_CREATE case and dropping it here
 523  * or in ll_file_release(), but I'm not sure that is desirable/necessary.
 524  */
 525 int ll_file_open(struct inode *inode, struct file *file)
 526 {
 527         struct ll_inode_info *lli = ll_i2info(inode);
 528         struct lookup_intent *it, oit = { .it_op = IT_OPEN,
 529                                           .it_flags = file->f_flags };
 530         struct obd_client_handle **och_p = NULL;
 531         __u64 *och_usecount = NULL;
 532         struct ll_file_data *fd;
 533         int rc = 0, opendir_set = 0;
 534         ENTRY;
 535
 536         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p), flags %o\n", inode->i_ino,
 537                inode->i_generation, inode, file->f_flags);
 538
 539         it = file->private_data; /* XXX: compat macro */
 540         file->private_data = NULL; /* prevent ll_local_open assertion */
 541
 542         fd = ll_file_data_get();
 543         if (fd == NULL)
 544                 GOTO(out_openerr, rc = -ENOMEM);
 545
 546         fd->fd_file = file;
 547         if (S_ISDIR(inode->i_mode)) {
 548                 spin_lock(&lli->lli_sa_lock);
 549                 if (lli->lli_opendir_key == NULL && lli->lli_sai == NULL &&
 550                     lli->lli_opendir_pid == 0) {
 551                         lli->lli_opendir_key = fd;
 552                         lli->lli_opendir_pid = cfs_curproc_pid();
 553                         opendir_set = 1;
 554                 }
 555                 spin_unlock(&lli->lli_sa_lock);
 556         }
 557
 558         if (inode->i_sb->s_root == file->f_dentry) {
 559                 LUSTRE_FPRIVATE(file) = fd;
 560                 RETURN(0);
 561         }
 562
 563         if (!it || !it->d.lustre.it_disposition) {
 564                 /* Convert f_flags into access mode. We cannot use file->f_mode,
 565                  * because everything but O_ACCMODE mask was stripped from
 566                  * there */
 567                 if ((oit.it_flags + 1) & O_ACCMODE)
 568                         oit.it_flags++;
 569                 if (file->f_flags & O_TRUNC)
 570                         oit.it_flags |= FMODE_WRITE;
 571
 572                 /* kernel only call f_op->open in dentry_open.  filp_open calls
 573                  * dentry_open after call to open_namei that checks permissions.
 574                  * Only nfsd_open call dentry_open directly without checking
 575                  * permissions and because of that this code below is safe. */
 576                 if (oit.it_flags & (FMODE_WRITE | FMODE_READ))
 577                         oit.it_flags |= MDS_OPEN_OWNEROVERRIDE;
 578
 579                 /* We do not want O_EXCL here, presumably we opened the file
 580                  * already? XXX - NFS implications? */
 581                 oit.it_flags &= ~O_EXCL;
 582
 583                 /* bug20584, if "it_flags" contains O_CREAT, the file will be
 584                  * created if necessary, then "IT_CREAT" should be set to keep
 585                  * consistent with it */
 586                 if (oit.it_flags & O_CREAT)
 587                         oit.it_op |= IT_CREAT;
 588
 589                 it = &oit;
 590         }
 591
 592 restart:
 593         /* Let's see if we have file open on MDS already. */
 594         if (it->it_flags & FMODE_WRITE) {
 595                 och_p = &lli->lli_mds_write_och;
 596                 och_usecount = &lli->lli_open_fd_write_count;
 597         } else if (it->it_flags & FMODE_EXEC) {
 598                 och_p = &lli->lli_mds_exec_och;
 599                 och_usecount = &lli->lli_open_fd_exec_count;
 600          } else {
 601                 och_p = &lli->lli_mds_read_och;
 602                 och_usecount = &lli->lli_open_fd_read_count;
 603         }
 604
 605         mutex_lock(&lli->lli_och_mutex);
 606         if (*och_p) { /* Open handle is present */
 607                 if (it_disposition(it, DISP_OPEN_OPEN)) {
 608                         /* Well, there's extra open request that we do not need,
 609                            let's close it somehow. This will decref request. */
 610                         rc = it_open_error(DISP_OPEN_OPEN, it);
 611                         if (rc) {
 612                                 mutex_unlock(&lli->lli_och_mutex);
 613                                 GOTO(out_openerr, rc);
 614                         }
 615
 616                         ll_release_openhandle(file->f_dentry, it);
 617                 }
 618                 (*och_usecount)++;
 619
 620                 rc = ll_local_open(file, it, fd, NULL);
 621                 if (rc) {
 622                         (*och_usecount)--;
 623                         mutex_unlock(&lli->lli_och_mutex);
 624                         GOTO(out_openerr, rc);
 625                 }
 626         } else {
 627                 LASSERT(*och_usecount == 0);
 628                 if (!it->d.lustre.it_disposition) {
 629                         /* We cannot just request lock handle now, new ELC code
 630                            means that one of other OPEN locks for this file
 631                            could be cancelled, and since blocking ast handler
 632                            would attempt to grab och_mutex as well, that would
 633                            result in a deadlock */
 634                         mutex_unlock(&lli->lli_och_mutex);
 635                         it->it_create_mode |= M_CHECK_STALE;
 636                         rc = ll_intent_file_open(file, NULL, 0, it);
 637                         it->it_create_mode &= ~M_CHECK_STALE;
 638                         if (rc)
 639                                 GOTO(out_openerr, rc);
 640
 641                         goto restart;
 642                 }
 643                 OBD_ALLOC(*och_p, sizeof (struct obd_client_handle));
 644                 if (!*och_p)
 645                         GOTO(out_och_free, rc = -ENOMEM);
 646
 647                 (*och_usecount)++;
 648
 649                 /* md_intent_lock() didn't get a request ref if there was an
 650                  * open error, so don't do cleanup on the request here
 651                  * (bug 3430) */
 652                 /* XXX (green): Should not we bail out on any error here, not
 653                  * just open error? */
 654                 rc = it_open_error(DISP_OPEN_OPEN, it);
 655                 if (rc)
 656                         GOTO(out_och_free, rc);
 657
 658                 LASSERT(it_disposition(it, DISP_ENQ_OPEN_REF));
 659
 660                 rc = ll_local_open(file, it, fd, *och_p);
 661                 if (rc)
 662                         GOTO(out_och_free, rc);
 663         }
 664         mutex_unlock(&lli->lli_och_mutex);
 665         fd = NULL;
 666
 667         /* Must do this outside lli_och_mutex lock to prevent deadlock where
 668            different kind of OPEN lock for this same inode gets cancelled
 669            by ldlm_cancel_lru */
 670         if (!S_ISREG(inode->i_mode))
 671                 GOTO(out_och_free, rc);
 672
 673         ll_capa_open(inode);
 674
 675         if (!lli->lli_has_smd) {
 676                 if (file->f_flags & O_LOV_DELAY_CREATE ||
 677                     !(file->f_mode & FMODE_WRITE)) {
 678                         CDEBUG(D_INODE, "object creation was delayed\n");
 679                         GOTO(out_och_free, rc);
 680                 }
 681         }
 682         file->f_flags &= ~O_LOV_DELAY_CREATE;
 683         GOTO(out_och_free, rc);
 684
 685 out_och_free:
 686         if (rc) {
 687                 if (och_p && *och_p) {
 688                         OBD_FREE(*och_p, sizeof (struct obd_client_handle));
 689                         *och_p = NULL; /* OBD_FREE writes some magic there */
 690                         (*och_usecount)--;
 691                 }
 692                 mutex_unlock(&lli->lli_och_mutex);
 693
 694 out_openerr:
 695                 if (opendir_set != 0)
 696                         ll_stop_statahead(inode, lli->lli_opendir_key);
 697                 if (fd != NULL)
 698                         ll_file_data_put(fd);
 699         } else {
 700                 ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_OPEN, 1);
 701         }
 702
 703         if (it && it_disposition(it, DISP_ENQ_OPEN_REF)) {
 704                 ptlrpc_req_finished(it->d.lustre.it_data);
 705                 it_clear_disposition(it, DISP_ENQ_OPEN_REF);
 706         }
 707
 708         return rc;
 709 }
 710
 711 static int ll_md_blocking_lease_ast(struct ldlm_lock *lock,
 712                         struct ldlm_lock_desc *desc, void *data, int flag)
 713 {
 714         int rc;
 715         struct lustre_handle lockh;
 716         ENTRY;
 717
 718         switch (flag) {
 719         case LDLM_CB_BLOCKING:
 720                 ldlm_lock2handle(lock, &lockh);
 721                 rc = ldlm_cli_cancel(&lockh, LCF_ASYNC);
 722                 if (rc < 0) {
 723                         CDEBUG(D_INODE, "ldlm_cli_cancel: %d\n", rc);
 724                         RETURN(rc);
 725                 }
 726                 break;
 727         case LDLM_CB_CANCELING:
 728                 /* do nothing */
 729                 break;
 730         }
 731         RETURN(0);
 732 }
 733
 734 /**
 735  * Acquire a lease and open the file.
 736  */
 737 struct obd_client_handle *ll_lease_open(struct inode *inode, struct file *file,
 738                                         fmode_t fmode)
 739 {
 740         struct lookup_intent it = { .it_op = IT_OPEN };
 741         struct ll_sb_info *sbi = ll_i2sbi(inode);
 742         struct md_op_data *op_data;
 743         struct ptlrpc_request *req;
 744         struct lustre_handle old_handle = { 0 };
 745         struct obd_client_handle *och = NULL;
 746         int rc;
 747         int rc2;
 748         ENTRY;
 749
 750         if (fmode != FMODE_WRITE && fmode != FMODE_READ)
 751                 RETURN(ERR_PTR(-EINVAL));
 752
 753         if (file != NULL) {
 754                 struct ll_inode_info *lli = ll_i2info(inode);
 755                 struct ll_file_data *fd = LUSTRE_FPRIVATE(file);
 756                 struct obd_client_handle **och_p;
 757                 __u64 *och_usecount;
 758
 759                 if (!(fmode & file->f_mode) || (file->f_mode & FMODE_EXEC))
 760                         RETURN(ERR_PTR(-EPERM));
 761
 762                 /* Get the openhandle of the file */
 763                 rc = -EBUSY;
 764                 mutex_lock(&lli->lli_och_mutex);
 765                 if (fd->fd_lease_och != NULL) {
 766                         mutex_unlock(&lli->lli_och_mutex);
 767                         RETURN(ERR_PTR(rc));
 768                 }
 769
 770                 if (fd->fd_och == NULL) {
 771                         if (file->f_mode & FMODE_WRITE) {
 772                                 LASSERT(lli->lli_mds_write_och != NULL);
 773                                 och_p = &lli->lli_mds_write_och;
 774                                 och_usecount = &lli->lli_open_fd_write_count;
 775                         } else {
 776                                 LASSERT(lli->lli_mds_read_och != NULL);
 777                                 och_p = &lli->lli_mds_read_och;
 778                                 och_usecount = &lli->lli_open_fd_read_count;
 779                         }
 780                         if (*och_usecount == 1) {
 781                                 fd->fd_och = *och_p;
 782                                 *och_p = NULL;
 783                                 *och_usecount = 0;
 784                                 rc = 0;
 785                         }
 786                 }
 787                 mutex_unlock(&lli->lli_och_mutex);
 788                 if (rc < 0) /* more than 1 opener */
 789                         RETURN(ERR_PTR(rc));
 790
 791                 LASSERT(fd->fd_och != NULL);
 792                 old_handle = fd->fd_och->och_fh;
 793         }
 794
 795         OBD_ALLOC_PTR(och);
 796         if (och == NULL)
 797                 RETURN(ERR_PTR(-ENOMEM));
 798
 799         op_data = ll_prep_md_op_data(NULL, inode, inode, NULL, 0, 0,
 800                                         LUSTRE_OPC_ANY, NULL);
 801         if (IS_ERR(op_data))
 802                 GOTO(out, rc = PTR_ERR(op_data));
 803
 804         /* To tell the MDT this openhandle is from the same owner */
 805         op_data->op_handle = old_handle;
 806
 807         it.it_flags = fmode | MDS_OPEN_LOCK | MDS_OPEN_BY_FID | MDS_OPEN_LEASE;
 808         rc = md_intent_lock(sbi->ll_md_exp, op_data, NULL, 0, &it, 0, &req,
 809                                 ll_md_blocking_lease_ast,
 810         /* LDLM_FL_NO_LRU: To not put the lease lock into LRU list, otherwise
 811          * it can be cancelled which may mislead applications that the lease is
 812          * broken;
 813          * LDLM_FL_EXCL: Set this flag so that it won't be matched by normal
 814          * open in ll_md_blocking_ast(). Otherwise as ll_md_blocking_lease_ast
 815          * doesn't deal with openhandle, so normal openhandle will be leaked. */
 816                                 LDLM_FL_NO_LRU | LDLM_FL_EXCL);
 817         ll_finish_md_op_data(op_data);
 818         if (req != NULL) {
 819                 ptlrpc_req_finished(req);
 820                 it_clear_disposition(&it, DISP_ENQ_COMPLETE);
 821         }
 822         if (rc < 0)
 823                 GOTO(out_release_it, rc);
 824
 825         if (it_disposition(&it, DISP_LOOKUP_NEG))
 826                 GOTO(out_release_it, rc = -ENOENT);
 827
 828         rc = it_open_error(DISP_OPEN_OPEN, &it);
 829         if (rc)
 830                 GOTO(out_release_it, rc);
 831
 832         LASSERT(it_disposition(&it, DISP_ENQ_OPEN_REF));
 833         ll_och_fill(sbi->ll_md_exp, &it, och);
 834
 835         if (!it_disposition(&it, DISP_OPEN_LEASE)) /* old server? */
 836                 GOTO(out_close, rc = -EOPNOTSUPP);
 837
 838         /* already get lease, handle lease lock */
 839         ll_set_lock_data(sbi->ll_md_exp, inode, &it, NULL);
 840         if (it.d.lustre.it_lock_mode == 0 ||
 841             it.d.lustre.it_lock_bits != MDS_INODELOCK_OPEN) {
 842                 /* open lock must return for lease */
 843                 CERROR(DFID "lease granted but no open lock, %d/%Lu.\n",
 844                         PFID(ll_inode2fid(inode)), it.d.lustre.it_lock_mode,
 845                         it.d.lustre.it_lock_bits);
 846                 GOTO(out_close, rc = -EPROTO);
 847         }
 848
 849         ll_intent_release(&it);
 850         RETURN(och);
 851
 852 out_close:
 853         rc2 = ll_close_inode_openhandle(sbi->ll_md_exp, inode, och);
 854         if (rc2)
 855                 CERROR("Close openhandle returned %d\n", rc2);
 856
 857         /* cancel open lock */
 858         if (it.d.lustre.it_lock_mode != 0) {
 859                 ldlm_lock_decref_and_cancel(&och->och_lease_handle,
 860                                                 it.d.lustre.it_lock_mode);
 861                 it.d.lustre.it_lock_mode = 0;
 862         }
 863 out_release_it:
 864         ll_intent_release(&it);
 865 out:
 866         OBD_FREE_PTR(och);
 867         RETURN(ERR_PTR(rc));
 868 }
 869 EXPORT_SYMBOL(ll_lease_open);
 870
 871 /**
 872  * Release lease and close the file.
 873  * It will check if the lease has ever broken.
 874  */
 875 int ll_lease_close(struct obd_client_handle *och, struct inode *inode,
 876                         bool *lease_broken)
 877 {
 878         struct ldlm_lock *lock;
 879         bool cancelled = true;
 880         int rc;
 881         ENTRY;
 882
 883         lock = ldlm_handle2lock(&och->och_lease_handle);
 884         if (lock != NULL) {
 885                 lock_res_and_lock(lock);
 886                 cancelled = ldlm_is_cancel(lock);
 887                 unlock_res_and_lock(lock);
 888                 ldlm_lock_put(lock);
 889         }
 890
 891         CDEBUG(D_INODE, "lease for "DFID" broken? %d\n",
 892                 PFID(&ll_i2info(inode)->lli_fid), cancelled);
 893
 894         if (!cancelled)
 895                 ldlm_cli_cancel(&och->och_lease_handle, 0);
 896         if (lease_broken != NULL)
 897                 *lease_broken = cancelled;
 898
 899         rc = ll_close_inode_openhandle(ll_i2sbi(inode)->ll_md_exp, inode, och);
 900         RETURN(rc);
 901 }
 902 EXPORT_SYMBOL(ll_lease_close);
 903
 904 /* Fills the obdo with the attributes for the lsm */
 905 static int ll_lsm_getattr(struct lov_stripe_md *lsm, struct obd_export *exp,
 906                           struct obd_capa *capa, struct obdo *obdo,
 907                           __u64 ioepoch, int sync)
 908 {
 909         struct ptlrpc_request_set *set;
 910         struct obd_info            oinfo = { { { 0 } } };
 911         int                        rc;
 912
 913         ENTRY;
 914
 915         LASSERT(lsm != NULL);
 916
 917         oinfo.oi_md = lsm;
 918         oinfo.oi_oa = obdo;
 919         oinfo.oi_oa->o_oi = lsm->lsm_oi;
 920         oinfo.oi_oa->o_mode = S_IFREG;
 921         oinfo.oi_oa->o_ioepoch = ioepoch;
 922         oinfo.oi_oa->o_valid = OBD_MD_FLID | OBD_MD_FLTYPE |
 923                                OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
 924                                OBD_MD_FLBLKSZ | OBD_MD_FLATIME |
 925                                OBD_MD_FLMTIME | OBD_MD_FLCTIME |
 926                                OBD_MD_FLGROUP | OBD_MD_FLEPOCH |
 927                                OBD_MD_FLDATAVERSION;
 928         oinfo.oi_capa = capa;
 929         if (sync) {
 930                 oinfo.oi_oa->o_valid |= OBD_MD_FLFLAGS;
 931                 oinfo.oi_oa->o_flags |= OBD_FL_SRVLOCK;
 932         }
 933
 934         set = ptlrpc_prep_set();
 935         if (set == NULL) {
 936                 CERROR("can't allocate ptlrpc set\n");
 937                 rc = -ENOMEM;
 938         } else {
 939                 rc = obd_getattr_async(exp, &oinfo, set);
 940                 if (rc == 0)
 941                         rc = ptlrpc_set_wait(set);
 942                 ptlrpc_set_destroy(set);
 943         }
 944         if (rc == 0)
 945                 oinfo.oi_oa->o_valid &= (OBD_MD_FLBLOCKS | OBD_MD_FLBLKSZ |
 946                                          OBD_MD_FLATIME | OBD_MD_FLMTIME |
 947                                          OBD_MD_FLCTIME | OBD_MD_FLSIZE |
 948                                          OBD_MD_FLDATAVERSION);
 949         RETURN(rc);
 950 }
 951
 952 /**
 953   * Performs the getattr on the inode and updates its fields.
 954   * If @sync != 0, perform the getattr under the server-side lock.
 955   */
 956 int ll_inode_getattr(struct inode *inode, struct obdo *obdo,
 957                      __u64 ioepoch, int sync)
 958 {
 959         struct obd_capa      *capa = ll_mdscapa_get(inode);
 960         struct lov_stripe_md *lsm;
 961         int rc;
 962         ENTRY;
 963
 964         lsm = ccc_inode_lsm_get(inode);
 965         rc = ll_lsm_getattr(lsm, ll_i2dtexp(inode),
 966                             capa, obdo, ioepoch, sync);
 967         capa_put(capa);
 968         if (rc == 0) {
 969                 struct ost_id *oi = lsm ? &lsm->lsm_oi : &obdo->o_oi;
 970
 971                 obdo_refresh_inode(inode, obdo, obdo->o_valid);
 972                 CDEBUG(D_INODE, "objid "DOSTID" size %llu, blocks %llu,"
 973                        " blksize %lu\n", POSTID(oi), i_size_read(inode),
 974                        (unsigned long long)inode->i_blocks,
 975                        (unsigned long)ll_inode_blksize(inode));
 976         }
 977         ccc_inode_lsm_put(inode, lsm);
 978         RETURN(rc);
 979 }
 980
 981 int ll_merge_lvb(const struct lu_env *env, struct inode *inode)
 982 {
 983         struct ll_inode_info *lli = ll_i2info(inode);
 984         struct cl_object *obj = lli->lli_clob;
 985         struct cl_attr *attr = ccc_env_thread_attr(env);
 986         struct ost_lvb lvb;
 987         int rc = 0;
 988
 989         ENTRY;
 990
 991         ll_inode_size_lock(inode);
 992         /* merge timestamps the most recently obtained from mds with
 993            timestamps obtained from osts */
 994         LTIME_S(inode->i_atime) = lli->lli_lvb.lvb_atime;
 995         LTIME_S(inode->i_mtime) = lli->lli_lvb.lvb_mtime;
 996         LTIME_S(inode->i_ctime) = lli->lli_lvb.lvb_ctime;
 997         inode_init_lvb(inode, &lvb);
 998
 999         cl_object_attr_lock(obj);
1000         rc = cl_object_attr_get(env, obj, attr);
1001         cl_object_attr_unlock(obj);
1002
1003         if (rc == 0) {
1004                 if (lvb.lvb_atime < attr->cat_atime)
1005                         lvb.lvb_atime = attr->cat_atime;
1006                 if (lvb.lvb_ctime < attr->cat_ctime)
1007                         lvb.lvb_ctime = attr->cat_ctime;
1008                 if (lvb.lvb_mtime < attr->cat_mtime)
1009                         lvb.lvb_mtime = attr->cat_mtime;
1010
1011                 CDEBUG(D_VFSTRACE, DFID" updating i_size "LPU64"\n",
1012                                 PFID(&lli->lli_fid), attr->cat_size);
1013                 cl_isize_write_nolock(inode, attr->cat_size);
1014
1015                 inode->i_blocks = attr->cat_blocks;
1016
1017                 LTIME_S(inode->i_mtime) = lvb.lvb_mtime;
1018                 LTIME_S(inode->i_atime) = lvb.lvb_atime;
1019                 LTIME_S(inode->i_ctime) = lvb.lvb_ctime;
1020         }
1021         ll_inode_size_unlock(inode);
1022
1023         RETURN(rc);
1024 }
1025
1026 int ll_glimpse_ioctl(struct ll_sb_info *sbi, struct lov_stripe_md *lsm,
1027                      lstat_t *st)
1028 {
1029         struct obdo obdo = { 0 };
1030         int rc;
1031
1032         rc = ll_lsm_getattr(lsm, sbi->ll_dt_exp, NULL, &obdo, 0, 0);
1033         if (rc == 0) {
1034                 st->st_size   = obdo.o_size;
1035                 st->st_blocks = obdo.o_blocks;
1036                 st->st_mtime  = obdo.o_mtime;
1037                 st->st_atime  = obdo.o_atime;
1038                 st->st_ctime  = obdo.o_ctime;
1039         }
1040         return rc;
1041 }
1042
1043 void ll_io_init(struct cl_io *io, const struct file *file, int write)
1044 {
1045         struct inode *inode = file->f_dentry->d_inode;
1046
1047         io->u.ci_rw.crw_nonblock = file->f_flags & O_NONBLOCK;
1048         if (write) {
1049                 io->u.ci_wr.wr_append = !!(file->f_flags & O_APPEND);
1050                 io->u.ci_wr.wr_sync = file->f_flags & O_SYNC ||
1051                                       file->f_flags & O_DIRECT ||
1052                                       IS_SYNC(inode);
1053         }
1054         io->ci_obj     = ll_i2info(inode)->lli_clob;
1055         io->ci_lockreq = CILR_MAYBE;
1056         if (ll_file_nolock(file)) {
1057                 io->ci_lockreq = CILR_NEVER;
1058                 io->ci_no_srvlock = 1;
1059         } else if (file->f_flags & O_APPEND) {
1060                 io->ci_lockreq = CILR_MANDATORY;
1061         }
1062 }
1063
1064 static ssize_t
1065 ll_file_io_generic(const struct lu_env *env, struct vvp_io_args *args,
1066                    struct file *file, enum cl_io_type iot,
1067                    loff_t *ppos, size_t count)
1068 {
1069         struct ll_inode_info *lli = ll_i2info(file->f_dentry->d_inode);
1070         struct ll_file_data  *fd  = LUSTRE_FPRIVATE(file);
1071         struct cl_io         *io;
1072         ssize_t               result;
1073         ENTRY;
1074
1075 restart:
1076         io = ccc_env_thread_io(env);
1077         ll_io_init(io, file, iot == CIT_WRITE);
1078
1079         if (cl_io_rw_init(env, io, iot, *ppos, count) == 0) {
1080                 struct vvp_io *vio = vvp_env_io(env);
1081                 struct ccc_io *cio = ccc_env_io(env);
1082                 int write_mutex_locked = 0;
1083
1084                 cio->cui_fd  = LUSTRE_FPRIVATE(file);
1085                 vio->cui_io_subtype = args->via_io_subtype;
1086
1087                 switch (vio->cui_io_subtype) {
1088                 case IO_NORMAL:
1089                         cio->cui_iov = args->u.normal.via_iov;
1090                         cio->cui_nrsegs = args->u.normal.via_nrsegs;
1091                         cio->cui_tot_nrsegs = cio->cui_nrsegs;
1092 #ifndef HAVE_FILE_WRITEV
1093                         cio->cui_iocb = args->u.normal.via_iocb;
1094 #endif
1095                         if ((iot == CIT_WRITE) &&
1096                             !(cio->cui_fd->fd_flags & LL_FILE_GROUP_LOCKED)) {
1097                                 if (mutex_lock_interruptible(&lli->
1098                                                                lli_write_mutex))
1099                                         GOTO(out, result = -ERESTARTSYS);
1100                                 write_mutex_locked = 1;
1101                         } else if (iot == CIT_READ) {
1102                                 down_read(&lli->lli_trunc_sem);
1103                         }
1104                         break;
1105                 case IO_SENDFILE:
1106                         vio->u.sendfile.cui_actor = args->u.sendfile.via_actor;
1107                         vio->u.sendfile.cui_target = args->u.sendfile.via_target;
1108                         break;
1109                 case IO_SPLICE:
1110                         vio->u.splice.cui_pipe = args->u.splice.via_pipe;
1111                         vio->u.splice.cui_flags = args->u.splice.via_flags;
1112                         break;
1113                 default:
1114                         CERROR("Unknow IO type - %u\n", vio->cui_io_subtype);
1115                         LBUG();
1116                 }
1117                 result = cl_io_loop(env, io);
1118                 if (write_mutex_locked)
1119                         mutex_unlock(&lli->lli_write_mutex);
1120                 else if (args->via_io_subtype == IO_NORMAL && iot == CIT_READ)
1121                         up_read(&lli->lli_trunc_sem);
1122         } else {
1123                 /* cl_io_rw_init() handled IO */
1124                 result = io->ci_result;
1125         }
1126
1127         if (io->ci_nob > 0) {
1128                 result = io->ci_nob;
1129                 *ppos = io->u.ci_wr.wr.crw_pos;
1130         }
1131         GOTO(out, result);
1132 out:
1133         cl_io_fini(env, io);
1134         /* If any bit been read/written (result != 0), we just return
1135          * short read/write instead of restart io. */
1136         if ((result == 0 || result == -ENODATA) && io->ci_need_restart) {
1137                 CDEBUG(D_VFSTRACE, "Restart %s on %s from %lld, count:%zd\n",
1138                        iot == CIT_READ ? "read" : "write",
1139                        file->f_dentry->d_name.name, *ppos, count);
1140                 LASSERTF(io->ci_nob == 0, "%zd", io->ci_nob);
1141                 goto restart;
1142         }
1143
1144         if (iot == CIT_READ) {
1145                 if (result >= 0)
1146                         ll_stats_ops_tally(ll_i2sbi(file->f_dentry->d_inode),
1147                                            LPROC_LL_READ_BYTES, result);
1148         } else if (iot == CIT_WRITE) {
1149                 if (result >= 0) {
1150                         ll_stats_ops_tally(ll_i2sbi(file->f_dentry->d_inode),
1151                                            LPROC_LL_WRITE_BYTES, result);
1152                         fd->fd_write_failed = false;
1153                 } else if (result != -ERESTARTSYS) {
1154                         fd->fd_write_failed = true;
1155                 }
1156         }
1157
1158         return result;
1159 }
1160
1161
1162 /*
1163  * XXX: exact copy from kernel code (__generic_file_aio_write_nolock)
1164  */
1165 static int ll_file_get_iov_count(const struct iovec *iov,
1166                                  unsigned long *nr_segs, size_t *count)
1167 {
1168         size_t cnt = 0;
1169         unsigned long seg;
1170
1171         for (seg = 0; seg < *nr_segs; seg++) {
1172                 const struct iovec *iv = &iov[seg];
1173
1174                 /*
1175                  * If any segment has a negative length, or the cumulative
1176                  * length ever wraps negative then return -EINVAL.
1177                  */
1178                 cnt += iv->iov_len;
1179                 if (unlikely((ssize_t)(cnt|iv->iov_len) < 0))
1180                         return -EINVAL;
1181                 if (access_ok(VERIFY_READ, iv->iov_base, iv->iov_len))
1182                         continue;
1183                 if (seg == 0)
1184                         return -EFAULT;
1185                 *nr_segs = seg;
1186                 cnt -= iv->iov_len;   /* This segment is no good */
1187                 break;
1188         }
1189         *count = cnt;
1190         return 0;
1191 }
1192
1193 #ifdef HAVE_FILE_READV
1194 static ssize_t ll_file_readv(struct file *file, const struct iovec *iov,
1195                               unsigned long nr_segs, loff_t *ppos)
1196 {
1197         struct lu_env      *env;
1198         struct vvp_io_args *args;
1199         size_t              count;
1200         ssize_t             result;
1201         int                 refcheck;
1202         ENTRY;
1203
1204         result = ll_file_get_iov_count(iov, &nr_segs, &count);
1205         if (result)
1206                 RETURN(result);
1207
1208         env = cl_env_get(&refcheck);
1209         if (IS_ERR(env))
1210                 RETURN(PTR_ERR(env));
1211
1212         args = vvp_env_args(env, IO_NORMAL);
1213         args->u.normal.via_iov = (struct iovec *)iov;
1214         args->u.normal.via_nrsegs = nr_segs;
1215
1216         result = ll_file_io_generic(env, args, file, CIT_READ, ppos, count);
1217         cl_env_put(env, &refcheck);
1218         RETURN(result);
1219 }
1220
1221 static ssize_t ll_file_read(struct file *file, char *buf, size_t count,
1222                             loff_t *ppos)
1223 {
1224         struct lu_env *env;
1225         struct iovec  *local_iov;
1226         ssize_t        result;
1227         int            refcheck;
1228         ENTRY;
1229
1230         env = cl_env_get(&refcheck);
1231         if (IS_ERR(env))
1232                 RETURN(PTR_ERR(env));
1233
1234         local_iov = &vvp_env_info(env)->vti_local_iov;
1235         local_iov->iov_base = (void __user *)buf;
1236         local_iov->iov_len = count;
1237         result = ll_file_readv(file, local_iov, 1, ppos);
1238         cl_env_put(env, &refcheck);
1239         RETURN(result);
1240 }
1241
1242 #else
1243 static ssize_t ll_file_aio_read(struct kiocb *iocb, const struct iovec *iov,
1244                                 unsigned long nr_segs, loff_t pos)
1245 {
1246         struct lu_env      *env;
1247         struct vvp_io_args *args;
1248         size_t              count;
1249         ssize_t             result;
1250         int                 refcheck;
1251         ENTRY;
1252
1253         result = ll_file_get_iov_count(iov, &nr_segs, &count);
1254         if (result)
1255                 RETURN(result);
1256
1257         env = cl_env_get(&refcheck);
1258         if (IS_ERR(env))
1259                 RETURN(PTR_ERR(env));
1260
1261         args = vvp_env_args(env, IO_NORMAL);
1262         args->u.normal.via_iov = (struct iovec *)iov;
1263         args->u.normal.via_nrsegs = nr_segs;
1264         args->u.normal.via_iocb = iocb;
1265
1266         result = ll_file_io_generic(env, args, iocb->ki_filp, CIT_READ,
1267                                     &iocb->ki_pos, count);
1268         cl_env_put(env, &refcheck);
1269         RETURN(result);
1270 }
1271
1272 static ssize_t ll_file_read(struct file *file, char *buf, size_t count,
1273                             loff_t *ppos)
1274 {
1275         struct lu_env *env;
1276         struct iovec  *local_iov;
1277         struct kiocb  *kiocb;
1278         ssize_t        result;
1279         int            refcheck;
1280         ENTRY;
1281
1282         env = cl_env_get(&refcheck);
1283         if (IS_ERR(env))
1284                 RETURN(PTR_ERR(env));
1285
1286         local_iov = &vvp_env_info(env)->vti_local_iov;
1287         kiocb = &vvp_env_info(env)->vti_kiocb;
1288         local_iov->iov_base = (void __user *)buf;
1289         local_iov->iov_len = count;
1290         init_sync_kiocb(kiocb, file);
1291         kiocb->ki_pos = *ppos;
1292         kiocb->ki_left = count;
1293
1294         result = ll_file_aio_read(kiocb, local_iov, 1, kiocb->ki_pos);
1295         *ppos = kiocb->ki_pos;
1296
1297         cl_env_put(env, &refcheck);
1298         RETURN(result);
1299 }
1300 #endif
1301
1302 /*
1303  * Write to a file (through the page cache).
1304  */
1305 #ifdef HAVE_FILE_WRITEV
1306 static ssize_t ll_file_writev(struct file *file, const struct iovec *iov,
1307                               unsigned long nr_segs, loff_t *ppos)
1308 {
1309         struct lu_env      *env;
1310         struct vvp_io_args *args;
1311         size_t              count;
1312         ssize_t             result;
1313         int                 refcheck;
1314         ENTRY;
1315
1316         result = ll_file_get_iov_count(iov, &nr_segs, &count);
1317         if (result)
1318                 RETURN(result);
1319
1320         env = cl_env_get(&refcheck);
1321         if (IS_ERR(env))
1322                 RETURN(PTR_ERR(env));
1323
1324         args = vvp_env_args(env, IO_NORMAL);
1325         args->u.normal.via_iov = (struct iovec *)iov;
1326         args->u.normal.via_nrsegs = nr_segs;
1327
1328         result = ll_file_io_generic(env, args, file, CIT_WRITE, ppos, count);
1329         cl_env_put(env, &refcheck);
1330         RETURN(result);
1331 }
1332
1333 static ssize_t ll_file_write(struct file *file, const char *buf, size_t count,
1334                              loff_t *ppos)
1335 {
1336         struct lu_env    *env;
1337         struct iovec     *local_iov;
1338         ssize_t           result;
1339         int               refcheck;
1340         ENTRY;
1341
1342         env = cl_env_get(&refcheck);
1343         if (IS_ERR(env))
1344                 RETURN(PTR_ERR(env));
1345
1346         local_iov = &vvp_env_info(env)->vti_local_iov;
1347         local_iov->iov_base = (void __user *)buf;
1348         local_iov->iov_len = count;
1349
1350         result = ll_file_writev(file, local_iov, 1, ppos);
1351         cl_env_put(env, &refcheck);
1352         RETURN(result);
1353 }
1354
1355 #else /* AIO stuff */
1356 static ssize_t ll_file_aio_write(struct kiocb *iocb, const struct iovec *iov,
1357                                  unsigned long nr_segs, loff_t pos)
1358 {
1359         struct lu_env      *env;
1360         struct vvp_io_args *args;
1361         size_t              count;
1362         ssize_t             result;
1363         int                 refcheck;
1364         ENTRY;
1365
1366         result = ll_file_get_iov_count(iov, &nr_segs, &count);
1367         if (result)
1368                 RETURN(result);
1369
1370         env = cl_env_get(&refcheck);
1371         if (IS_ERR(env))
1372                 RETURN(PTR_ERR(env));
1373
1374         args = vvp_env_args(env, IO_NORMAL);
1375         args->u.normal.via_iov = (struct iovec *)iov;
1376         args->u.normal.via_nrsegs = nr_segs;
1377         args->u.normal.via_iocb = iocb;
1378
1379         result = ll_file_io_generic(env, args, iocb->ki_filp, CIT_WRITE,
1380                                   &iocb->ki_pos, count);
1381         cl_env_put(env, &refcheck);
1382         RETURN(result);
1383 }
1384
1385 static ssize_t ll_file_write(struct file *file, const char *buf, size_t count,
1386                              loff_t *ppos)
1387 {
1388         struct lu_env *env;
1389         struct iovec  *local_iov;
1390         struct kiocb  *kiocb;
1391         ssize_t        result;
1392         int            refcheck;
1393         ENTRY;
1394
1395         env = cl_env_get(&refcheck);
1396         if (IS_ERR(env))
1397                 RETURN(PTR_ERR(env));
1398
1399         local_iov = &vvp_env_info(env)->vti_local_iov;
1400         kiocb = &vvp_env_info(env)->vti_kiocb;
1401         local_iov->iov_base = (void __user *)buf;
1402         local_iov->iov_len = count;
1403         init_sync_kiocb(kiocb, file);
1404         kiocb->ki_pos = *ppos;
1405         kiocb->ki_left = count;
1406
1407         result = ll_file_aio_write(kiocb, local_iov, 1, kiocb->ki_pos);
1408         *ppos = kiocb->ki_pos;
1409
1410         cl_env_put(env, &refcheck);
1411         RETURN(result);
1412 }
1413 #endif
1414
1415 /*
1416  * Send file content (through pagecache) somewhere with helper
1417  */
1418 static ssize_t ll_file_splice_read(struct file *in_file, loff_t *ppos,
1419                                    struct pipe_inode_info *pipe, size_t count,
1420                                    unsigned int flags)
1421 {
1422         struct lu_env      *env;
1423         struct vvp_io_args *args;
1424         ssize_t             result;
1425         int                 refcheck;
1426         ENTRY;
1427
1428         env = cl_env_get(&refcheck);
1429         if (IS_ERR(env))
1430                 RETURN(PTR_ERR(env));
1431
1432         args = vvp_env_args(env, IO_SPLICE);
1433         args->u.splice.via_pipe = pipe;
1434         args->u.splice.via_flags = flags;
1435
1436         result = ll_file_io_generic(env, args, in_file, CIT_READ, ppos, count);
1437         cl_env_put(env, &refcheck);
1438         RETURN(result);
1439 }
1440
1441 static int ll_lov_recreate(struct inode *inode, struct ost_id *oi,
1442                            obd_count ost_idx)
1443 {
1444         struct obd_export *exp = ll_i2dtexp(inode);
1445         struct obd_trans_info oti = { 0 };
1446         struct obdo *oa = NULL;
1447         int lsm_size;
1448         int rc = 0;
1449         struct lov_stripe_md *lsm = NULL, *lsm2;
1450         ENTRY;
1451
1452         OBDO_ALLOC(oa);
1453         if (oa == NULL)
1454                 RETURN(-ENOMEM);
1455
1456         lsm = ccc_inode_lsm_get(inode);
1457         if (!lsm_has_objects(lsm))
1458                 GOTO(out, rc = -ENOENT);
1459
1460         lsm_size = sizeof(*lsm) + (sizeof(struct lov_oinfo) *
1461                    (lsm->lsm_stripe_count));
1462
1463         OBD_ALLOC_LARGE(lsm2, lsm_size);
1464         if (lsm2 == NULL)
1465                 GOTO(out, rc = -ENOMEM);
1466
1467         oa->o_oi = *oi;
1468         oa->o_nlink = ost_idx;
1469         oa->o_flags |= OBD_FL_RECREATE_OBJS;
1470         oa->o_valid = OBD_MD_FLID | OBD_MD_FLFLAGS | OBD_MD_FLGROUP;
1471         obdo_from_inode(oa, inode, OBD_MD_FLTYPE | OBD_MD_FLATIME |
1472                                    OBD_MD_FLMTIME | OBD_MD_FLCTIME);
1473         obdo_set_parent_fid(oa, &ll_i2info(inode)->lli_fid);
1474         memcpy(lsm2, lsm, lsm_size);
1475         ll_inode_size_lock(inode);
1476         rc = obd_create(NULL, exp, oa, &lsm2, &oti);
1477         ll_inode_size_unlock(inode);
1478
1479         OBD_FREE_LARGE(lsm2, lsm_size);
1480         GOTO(out, rc);
1481 out:
1482         ccc_inode_lsm_put(inode, lsm);
1483         OBDO_FREE(oa);
1484         return rc;
1485 }
1486
1487 static int ll_lov_recreate_obj(struct inode *inode, unsigned long arg)
1488 {
1489         struct ll_recreate_obj ucreat;
1490         struct ost_id           oi;
1491         ENTRY;
1492
1493         if (!cfs_capable(CFS_CAP_SYS_ADMIN))
1494                 RETURN(-EPERM);
1495
1496         if (copy_from_user(&ucreat, (struct ll_recreate_obj *)arg,
1497                            sizeof(ucreat)))
1498                 RETURN(-EFAULT);
1499
1500         ostid_set_seq_mdt0(&oi);
1501         ostid_set_id(&oi, ucreat.lrc_id);
1502         RETURN(ll_lov_recreate(inode, &oi, ucreat.lrc_ost_idx));
1503 }
1504
1505 static int ll_lov_recreate_fid(struct inode *inode, unsigned long arg)
1506 {
1507         struct lu_fid   fid;
1508         struct ost_id   oi;
1509         obd_count       ost_idx;
1510         ENTRY;
1511
1512         if (!cfs_capable(CFS_CAP_SYS_ADMIN))
1513                 RETURN(-EPERM);
1514
1515         if (copy_from_user(&fid, (struct lu_fid *)arg, sizeof(fid)))
1516                 RETURN(-EFAULT);
1517
1518         fid_to_ostid(&fid, &oi);
1519         ost_idx = (fid_seq(&fid) >> 16) & 0xffff;
1520         RETURN(ll_lov_recreate(inode, &oi, ost_idx));
1521 }
1522
1523 int ll_lov_setstripe_ea_info(struct inode *inode, struct file *file,
1524                              int flags, struct lov_user_md *lum, int lum_size)
1525 {
1526         struct lov_stripe_md *lsm = NULL;
1527         struct lookup_intent oit = {.it_op = IT_OPEN, .it_flags = flags};
1528         int rc = 0;
1529         ENTRY;
1530
1531         lsm = ccc_inode_lsm_get(inode);
1532         if (lsm != NULL) {
1533                 ccc_inode_lsm_put(inode, lsm);
1534                 CDEBUG(D_IOCTL, "stripe already exists for ino %lu\n",
1535                        inode->i_ino);
1536                 RETURN(-EEXIST);
1537         }
1538
1539         ll_inode_size_lock(inode);
1540         rc = ll_intent_file_open(file, lum, lum_size, &oit);
1541         if (rc)
1542                 GOTO(out, rc);
1543         rc = oit.d.lustre.it_status;
1544         if (rc < 0)
1545                 GOTO(out_req_free, rc);
1546
1547         ll_release_openhandle(file->f_dentry, &oit);
1548
1549  out:
1550         ll_inode_size_unlock(inode);
1551         ll_intent_release(&oit);
1552         ccc_inode_lsm_put(inode, lsm);
1553         RETURN(rc);
1554 out_req_free:
1555         ptlrpc_req_finished((struct ptlrpc_request *) oit.d.lustre.it_data);
1556         goto out;
1557 }
1558
1559 int ll_lov_getstripe_ea_info(struct inode *inode, const char *filename,
1560                              struct lov_mds_md **lmmp, int *lmm_size,
1561                              struct ptlrpc_request **request)
1562 {
1563         struct ll_sb_info *sbi = ll_i2sbi(inode);
1564         struct mdt_body  *body;
1565         struct lov_mds_md *lmm = NULL;
1566         struct ptlrpc_request *req = NULL;
1567         struct md_op_data *op_data;
1568         int rc, lmmsize;
1569
1570         rc = ll_get_max_mdsize(sbi, &lmmsize);
1571         if (rc)
1572                 RETURN(rc);
1573
1574         op_data = ll_prep_md_op_data(NULL, inode, NULL, filename,
1575                                      strlen(filename), lmmsize,
1576                                      LUSTRE_OPC_ANY, NULL);
1577         if (IS_ERR(op_data))
1578                 RETURN(PTR_ERR(op_data));
1579
1580         op_data->op_valid = OBD_MD_FLEASIZE | OBD_MD_FLDIREA;
1581         rc = md_getattr_name(sbi->ll_md_exp, op_data, &req);
1582         ll_finish_md_op_data(op_data);
1583         if (rc < 0) {
1584                 CDEBUG(D_INFO, "md_getattr_name failed "
1585                        "on %s: rc %d\n", filename, rc);
1586                 GOTO(out, rc);
1587         }
1588
1589         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
1590         LASSERT(body != NULL); /* checked by mdc_getattr_name */
1591
1592         lmmsize = body->eadatasize;
1593
1594         if (!(body->valid & (OBD_MD_FLEASIZE | OBD_MD_FLDIREA)) ||
1595                         lmmsize == 0) {
1596                 GOTO(out, rc = -ENODATA);
1597         }
1598
1599         lmm = req_capsule_server_sized_get(&req->rq_pill, &RMF_MDT_MD, lmmsize);
1600         LASSERT(lmm != NULL);
1601
1602         if ((lmm->lmm_magic != cpu_to_le32(LOV_MAGIC_V1)) &&
1603             (lmm->lmm_magic != cpu_to_le32(LOV_MAGIC_V3))) {
1604                 GOTO(out, rc = -EPROTO);
1605         }
1606
1607         /*
1608          * This is coming from the MDS, so is probably in
1609          * little endian.  We convert it to host endian before
1610          * passing it to userspace.
1611          */
1612         if (LOV_MAGIC != cpu_to_le32(LOV_MAGIC)) {
1613                 int stripe_count;
1614
1615                 stripe_count = le16_to_cpu(lmm->lmm_stripe_count);
1616                 if (le32_to_cpu(lmm->lmm_pattern) & LOV_PATTERN_F_RELEASED)
1617                         stripe_count = 0;
1618
1619                 /* if function called for directory - we should
1620                  * avoid swab not existent lsm objects */
1621                 if (lmm->lmm_magic == cpu_to_le32(LOV_MAGIC_V1)) {
1622                         lustre_swab_lov_user_md_v1((struct lov_user_md_v1 *)lmm);
1623                         if (S_ISREG(body->mode))
1624                                 lustre_swab_lov_user_md_objects(
1625                                  ((struct lov_user_md_v1 *)lmm)->lmm_objects,
1626                                  stripe_count);
1627                 } else if (lmm->lmm_magic == cpu_to_le32(LOV_MAGIC_V3)) {
1628                         lustre_swab_lov_user_md_v3((struct lov_user_md_v3 *)lmm);
1629                         if (S_ISREG(body->mode))
1630                                 lustre_swab_lov_user_md_objects(
1631                                  ((struct lov_user_md_v3 *)lmm)->lmm_objects,
1632                                  stripe_count);
1633                 }
1634         }
1635
1636 out:
1637         *lmmp = lmm;
1638         *lmm_size = lmmsize;
1639         *request = req;
1640         return rc;
1641 }
1642
1643 static int ll_lov_setea(struct inode *inode, struct file *file,
1644                             unsigned long arg)
1645 {
1646         int                      flags = MDS_OPEN_HAS_OBJS | FMODE_WRITE;
1647         struct lov_user_md      *lump;
1648         int                      lum_size = sizeof(struct lov_user_md) +
1649                                             sizeof(struct lov_user_ost_data);
1650         int                      rc;
1651         ENTRY;
1652
1653         if (!cfs_capable(CFS_CAP_SYS_ADMIN))
1654                 RETURN(-EPERM);
1655
1656         OBD_ALLOC_LARGE(lump, lum_size);
1657         if (lump == NULL)
1658                 RETURN(-ENOMEM);
1659
1660         if (copy_from_user(lump, (struct lov_user_md  *)arg, lum_size)) {
1661                 OBD_FREE_LARGE(lump, lum_size);
1662                 RETURN(-EFAULT);
1663         }
1664
1665         rc = ll_lov_setstripe_ea_info(inode, file, flags, lump, lum_size);
1666
1667         OBD_FREE_LARGE(lump, lum_size);
1668         RETURN(rc);
1669 }
1670
1671 static int ll_lov_setstripe(struct inode *inode, struct file *file,
1672                             unsigned long arg)
1673 {
1674         struct lov_user_md_v3    lumv3;
1675         struct lov_user_md_v1   *lumv1 = (struct lov_user_md_v1 *)&lumv3;
1676         struct lov_user_md_v1   *lumv1p = (struct lov_user_md_v1 *)arg;
1677         struct lov_user_md_v3   *lumv3p = (struct lov_user_md_v3 *)arg;
1678         int                      lum_size, rc;
1679         int                      flags = FMODE_WRITE;
1680         ENTRY;
1681
1682         /* first try with v1 which is smaller than v3 */
1683         lum_size = sizeof(struct lov_user_md_v1);
1684         if (copy_from_user(lumv1, lumv1p, lum_size))
1685                 RETURN(-EFAULT);
1686
1687         if (lumv1->lmm_magic == LOV_USER_MAGIC_V3) {
1688                 lum_size = sizeof(struct lov_user_md_v3);
1689                 if (copy_from_user(&lumv3, lumv3p, lum_size))
1690                         RETURN(-EFAULT);
1691         }
1692
1693         rc = ll_lov_setstripe_ea_info(inode, file, flags, lumv1, lum_size);
1694         if (rc == 0) {
1695                 struct lov_stripe_md *lsm;
1696                 __u32 gen;
1697
1698                 put_user(0, &lumv1p->lmm_stripe_count);
1699
1700                 ll_layout_refresh(inode, &gen);
1701                 lsm = ccc_inode_lsm_get(inode);
1702                 rc = obd_iocontrol(LL_IOC_LOV_GETSTRIPE, ll_i2dtexp(inode),
1703                                    0, lsm, (void *)arg);
1704                 ccc_inode_lsm_put(inode, lsm);
1705         }
1706         RETURN(rc);
1707 }
1708
1709 static int ll_lov_getstripe(struct inode *inode, unsigned long arg)
1710 {
1711         struct lov_stripe_md *lsm;
1712         int rc = -ENODATA;
1713         ENTRY;
1714
1715         lsm = ccc_inode_lsm_get(inode);
1716         if (lsm != NULL)
1717                 rc = obd_iocontrol(LL_IOC_LOV_GETSTRIPE, ll_i2dtexp(inode), 0,
1718                                    lsm, (void *)arg);
1719         ccc_inode_lsm_put(inode, lsm);
1720         RETURN(rc);
1721 }
1722
1723 int ll_get_grouplock(struct inode *inode, struct file *file, unsigned long arg)
1724 {
1725         struct ll_inode_info   *lli = ll_i2info(inode);
1726         struct ll_file_data    *fd = LUSTRE_FPRIVATE(file);
1727         struct ccc_grouplock    grouplock;
1728         int                     rc;
1729         ENTRY;
1730
1731         if (ll_file_nolock(file))
1732                 RETURN(-EOPNOTSUPP);
1733
1734         spin_lock(&lli->lli_lock);
1735         if (fd->fd_flags & LL_FILE_GROUP_LOCKED) {
1736                 CWARN("group lock already existed with gid %lu\n",
1737                       fd->fd_grouplock.cg_gid);
1738                 spin_unlock(&lli->lli_lock);
1739                 RETURN(-EINVAL);
1740         }
1741         LASSERT(fd->fd_grouplock.cg_lock == NULL);
1742         spin_unlock(&lli->lli_lock);
1743
1744         rc = cl_get_grouplock(cl_i2info(inode)->lli_clob,
1745                               arg, (file->f_flags & O_NONBLOCK), &grouplock);
1746         if (rc)
1747                 RETURN(rc);
1748
1749         spin_lock(&lli->lli_lock);
1750         if (fd->fd_flags & LL_FILE_GROUP_LOCKED) {
1751                 spin_unlock(&lli->lli_lock);
1752                 CERROR("another thread just won the race\n");
1753                 cl_put_grouplock(&grouplock);
1754                 RETURN(-EINVAL);
1755         }
1756
1757         fd->fd_flags |= LL_FILE_GROUP_LOCKED;
1758         fd->fd_grouplock = grouplock;
1759         spin_unlock(&lli->lli_lock);
1760
1761         CDEBUG(D_INFO, "group lock %lu obtained\n", arg);
1762         RETURN(0);
1763 }
1764
1765 int ll_put_grouplock(struct inode *inode, struct file *file, unsigned long arg)
1766 {
1767         struct ll_inode_info   *lli = ll_i2info(inode);
1768         struct ll_file_data    *fd = LUSTRE_FPRIVATE(file);
1769         struct ccc_grouplock    grouplock;
1770         ENTRY;
1771
1772         spin_lock(&lli->lli_lock);
1773         if (!(fd->fd_flags & LL_FILE_GROUP_LOCKED)) {
1774                 spin_unlock(&lli->lli_lock);
1775                 CWARN("no group lock held\n");
1776                 RETURN(-EINVAL);
1777         }
1778         LASSERT(fd->fd_grouplock.cg_lock != NULL);
1779
1780         if (fd->fd_grouplock.cg_gid != arg) {
1781                 CWARN("group lock %lu doesn't match current id %lu\n",
1782                        arg, fd->fd_grouplock.cg_gid);
1783                 spin_unlock(&lli->lli_lock);
1784                 RETURN(-EINVAL);
1785         }
1786
1787         grouplock = fd->fd_grouplock;
1788         memset(&fd->fd_grouplock, 0, sizeof(fd->fd_grouplock));
1789         fd->fd_flags &= ~LL_FILE_GROUP_LOCKED;
1790         spin_unlock(&lli->lli_lock);
1791
1792         cl_put_grouplock(&grouplock);
1793         CDEBUG(D_INFO, "group lock %lu released\n", arg);
1794         RETURN(0);
1795 }
1796
1797 /**
1798  * Close inode open handle
1799  *
1800  * \param dentry [in]     dentry which contains the inode
1801  * \param it     [in,out] intent which contains open info and result
1802  *
1803  * \retval 0     success
1804  * \retval <0    failure
1805  */
1806 int ll_release_openhandle(struct dentry *dentry, struct lookup_intent *it)
1807 {
1808         struct inode *inode = dentry->d_inode;
1809         struct obd_client_handle *och;
1810         int rc;
1811         ENTRY;
1812
1813         LASSERT(inode);
1814
1815         /* Root ? Do nothing. */
1816         if (dentry->d_inode->i_sb->s_root == dentry)
1817                 RETURN(0);
1818
1819         /* No open handle to close? Move away */
1820         if (!it_disposition(it, DISP_OPEN_OPEN))
1821                 RETURN(0);
1822
1823         LASSERT(it_open_error(DISP_OPEN_OPEN, it) == 0);
1824
1825         OBD_ALLOC(och, sizeof(*och));
1826         if (!och)
1827                 GOTO(out, rc = -ENOMEM);
1828
1829         ll_och_fill(ll_i2sbi(inode)->ll_md_exp, it, och);
1830
1831         rc = ll_close_inode_openhandle(ll_i2sbi(inode)->ll_md_exp,
1832                                        inode, och);
1833  out:
1834         /* this one is in place of ll_file_open */
1835         if (it_disposition(it, DISP_ENQ_OPEN_REF)) {
1836                 ptlrpc_req_finished(it->d.lustre.it_data);
1837                 it_clear_disposition(it, DISP_ENQ_OPEN_REF);
1838         }
1839         RETURN(rc);
1840 }
1841
1842 /**
1843  * Get size for inode for which FIEMAP mapping is requested.
1844  * Make the FIEMAP get_info call and returns the result.
1845  */
1846 int ll_do_fiemap(struct inode *inode, struct ll_user_fiemap *fiemap,
1847               int num_bytes)
1848 {
1849         struct obd_export *exp = ll_i2dtexp(inode);
1850         struct lov_stripe_md *lsm = NULL;
1851         struct ll_fiemap_info_key fm_key = { .name = KEY_FIEMAP, };
1852         int vallen = num_bytes;
1853         int rc;
1854         ENTRY;
1855
1856         /* Checks for fiemap flags */
1857         if (fiemap->fm_flags & ~LUSTRE_FIEMAP_FLAGS_COMPAT) {
1858                 fiemap->fm_flags &= ~LUSTRE_FIEMAP_FLAGS_COMPAT;
1859                 return -EBADR;
1860         }
1861
1862         /* Check for FIEMAP_FLAG_SYNC */
1863         if (fiemap->fm_flags & FIEMAP_FLAG_SYNC) {
1864                 rc = filemap_fdatawrite(inode->i_mapping);
1865                 if (rc)
1866                         return rc;
1867         }
1868
1869         lsm = ccc_inode_lsm_get(inode);
1870         if (lsm == NULL)
1871                 return -ENOENT;
1872
1873         /* If the stripe_count > 1 and the application does not understand
1874          * DEVICE_ORDER flag, then it cannot interpret the extents correctly.
1875          */
1876         if (lsm->lsm_stripe_count > 1 &&
1877             !(fiemap->fm_flags & FIEMAP_FLAG_DEVICE_ORDER))
1878                 GOTO(out, rc = -EOPNOTSUPP);
1879
1880         fm_key.oa.o_oi = lsm->lsm_oi;
1881         fm_key.oa.o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
1882
1883         obdo_from_inode(&fm_key.oa, inode, OBD_MD_FLSIZE);
1884         obdo_set_parent_fid(&fm_key.oa, &ll_i2info(inode)->lli_fid);
1885         /* If filesize is 0, then there would be no objects for mapping */
1886         if (fm_key.oa.o_size == 0) {
1887                 fiemap->fm_mapped_extents = 0;
1888                 GOTO(out, rc = 0);
1889         }
1890
1891         memcpy(&fm_key.fiemap, fiemap, sizeof(*fiemap));
1892
1893         rc = obd_get_info(NULL, exp, sizeof(fm_key), &fm_key, &vallen,
1894                           fiemap, lsm);
1895         if (rc)
1896                 CERROR("obd_get_info failed: rc = %d\n", rc);
1897
1898 out:
1899         ccc_inode_lsm_put(inode, lsm);
1900         RETURN(rc);
1901 }
1902
1903 int ll_fid2path(struct inode *inode, void *arg)
1904 {
1905         struct obd_export       *exp = ll_i2mdexp(inode);
1906         struct getinfo_fid2path *gfout, *gfin;
1907         int                      outsize, rc;
1908         ENTRY;
1909
1910         if (!cfs_capable(CFS_CAP_DAC_READ_SEARCH) &&
1911             !(ll_i2sbi(inode)->ll_flags & LL_SBI_USER_FID2PATH))
1912                 RETURN(-EPERM);
1913
1914         /* Need to get the buflen */
1915         OBD_ALLOC_PTR(gfin);
1916         if (gfin == NULL)
1917                 RETURN(-ENOMEM);
1918         if (copy_from_user(gfin, arg, sizeof(*gfin))) {
1919                 OBD_FREE_PTR(gfin);
1920                 RETURN(-EFAULT);
1921         }
1922
1923         outsize = sizeof(*gfout) + gfin->gf_pathlen;
1924         OBD_ALLOC(gfout, outsize);
1925         if (gfout == NULL) {
1926                 OBD_FREE_PTR(gfin);
1927                 RETURN(-ENOMEM);
1928         }
1929         memcpy(gfout, gfin, sizeof(*gfout));
1930         OBD_FREE_PTR(gfin);
1931
1932         /* Call mdc_iocontrol */
1933         rc = obd_iocontrol(OBD_IOC_FID2PATH, exp, outsize, gfout, NULL);
1934         if (rc)
1935                 GOTO(gf_free, rc);
1936
1937         if (copy_to_user(arg, gfout, outsize))
1938                 rc = -EFAULT;
1939
1940 gf_free:
1941         OBD_FREE(gfout, outsize);
1942         RETURN(rc);
1943 }
1944
1945 static int ll_ioctl_fiemap(struct inode *inode, unsigned long arg)
1946 {
1947         struct ll_user_fiemap *fiemap_s;
1948         size_t num_bytes, ret_bytes;
1949         unsigned int extent_count;
1950         int rc = 0;
1951
1952         /* Get the extent count so we can calculate the size of
1953          * required fiemap buffer */
1954         if (get_user(extent_count,
1955             &((struct ll_user_fiemap __user *)arg)->fm_extent_count))
1956                 RETURN(-EFAULT);
1957         num_bytes = sizeof(*fiemap_s) + (extent_count *
1958                                          sizeof(struct ll_fiemap_extent));
1959
1960         OBD_ALLOC_LARGE(fiemap_s, num_bytes);
1961         if (fiemap_s == NULL)
1962                 RETURN(-ENOMEM);
1963
1964         /* get the fiemap value */
1965         if (copy_from_user(fiemap_s, (struct ll_user_fiemap __user *)arg,
1966                            sizeof(*fiemap_s)))
1967                 GOTO(error, rc = -EFAULT);
1968
1969         /* If fm_extent_count is non-zero, read the first extent since
1970          * it is used to calculate end_offset and device from previous
1971          * fiemap call. */
1972         if (extent_count) {
1973                 if (copy_from_user(&fiemap_s->fm_extents[0],
1974                     (char __user *)arg + sizeof(*fiemap_s),
1975                     sizeof(struct ll_fiemap_extent)))
1976                         GOTO(error, rc = -EFAULT);
1977         }
1978
1979         rc = ll_do_fiemap(inode, fiemap_s, num_bytes);
1980         if (rc)
1981                 GOTO(error, rc);
1982
1983         ret_bytes = sizeof(struct ll_user_fiemap);
1984
1985         if (extent_count != 0)
1986                 ret_bytes += (fiemap_s->fm_mapped_extents *
1987                                  sizeof(struct ll_fiemap_extent));
1988
1989         if (copy_to_user((void *)arg, fiemap_s, ret_bytes))
1990                 rc = -EFAULT;
1991
1992 error:
1993         OBD_FREE_LARGE(fiemap_s, num_bytes);
1994         RETURN(rc);
1995 }
1996
1997 /*
1998  * Read the data_version for inode.
1999  *
2000  * This value is computed using stripe object version on OST.
2001  * Version is computed using server side locking.
2002  *
2003  * @param extent_lock  Take extent lock. Not needed if a process is already
2004  *                     holding the OST object group locks.
2005  */
2006 int ll_data_version(struct inode *inode, __u64 *data_version,
2007                     int extent_lock)
2008 {
2009         struct lov_stripe_md    *lsm = NULL;
2010         struct ll_sb_info       *sbi = ll_i2sbi(inode);
2011         struct obdo             *obdo = NULL;
2012         int                      rc;
2013         ENTRY;
2014
2015         /* If no stripe, we consider version is 0. */
2016         lsm = ccc_inode_lsm_get(inode);
2017         if (!lsm_has_objects(lsm)) {
2018                 *data_version = 0;
2019                 CDEBUG(D_INODE, "No object for inode\n");
2020                 GOTO(out, rc = 0);
2021         }
2022
2023         OBD_ALLOC_PTR(obdo);
2024         if (obdo == NULL)
2025                 GOTO(out, rc = -ENOMEM);
2026
2027         rc = ll_lsm_getattr(lsm, sbi->ll_dt_exp, NULL, obdo, 0, extent_lock);
2028         if (rc == 0) {
2029                 if (!(obdo->o_valid & OBD_MD_FLDATAVERSION))
2030                         rc = -EOPNOTSUPP;
2031                 else
2032                         *data_version = obdo->o_data_version;
2033         }
2034
2035         OBD_FREE_PTR(obdo);
2036         EXIT;
2037 out:
2038         ccc_inode_lsm_put(inode, lsm);
2039         RETURN(rc);
2040 }
2041
2042 struct ll_swap_stack {
2043         struct iattr             ia1, ia2;
2044         __u64                    dv1, dv2;
2045         struct inode            *inode1, *inode2;
2046         bool                     check_dv1, check_dv2;
2047 };
2048
2049 static int ll_swap_layouts(struct file *file1, struct file *file2,
2050                            struct lustre_swap_layouts *lsl)
2051 {
2052         struct mdc_swap_layouts  msl;
2053         struct md_op_data       *op_data;
2054         __u32                    gid;
2055         __u64                    dv;
2056         struct ll_swap_stack    *llss = NULL;
2057         int                      rc;
2058
2059         OBD_ALLOC_PTR(llss);
2060         if (llss == NULL)
2061                 RETURN(-ENOMEM);
2062
2063         llss->inode1 = file1->f_dentry->d_inode;
2064         llss->inode2 = file2->f_dentry->d_inode;
2065
2066         if (!S_ISREG(llss->inode2->i_mode))
2067                 GOTO(free, rc = -EINVAL);
2068
2069         if (inode_permission(llss->inode1, MAY_WRITE) ||
2070             inode_permission(llss->inode2, MAY_WRITE))
2071                 GOTO(free, rc = -EPERM);
2072
2073         if (llss->inode2->i_sb != llss->inode1->i_sb)
2074                 GOTO(free, rc = -EXDEV);
2075
2076         /* we use 2 bool because it is easier to swap than 2 bits */
2077         if (lsl->sl_flags & SWAP_LAYOUTS_CHECK_DV1)
2078                 llss->check_dv1 = true;
2079
2080         if (lsl->sl_flags & SWAP_LAYOUTS_CHECK_DV2)
2081                 llss->check_dv2 = true;
2082
2083         /* we cannot use lsl->sl_dvX directly because we may swap them */
2084         llss->dv1 = lsl->sl_dv1;
2085         llss->dv2 = lsl->sl_dv2;
2086
2087         rc = lu_fid_cmp(ll_inode2fid(llss->inode1), ll_inode2fid(llss->inode2));
2088         if (rc == 0) /* same file, done! */
2089                 GOTO(free, rc = 0);
2090
2091         if (rc < 0) { /* sequentialize it */
2092                 swap(llss->inode1, llss->inode2);
2093                 swap(file1, file2);
2094                 swap(llss->dv1, llss->dv2);
2095                 swap(llss->check_dv1, llss->check_dv2);
2096         }
2097
2098         gid = lsl->sl_gid;
2099         if (gid != 0) { /* application asks to flush dirty cache */
2100                 rc = ll_get_grouplock(llss->inode1, file1, gid);
2101                 if (rc < 0)
2102                         GOTO(free, rc);
2103
2104                 rc = ll_get_grouplock(llss->inode2, file2, gid);
2105                 if (rc < 0) {
2106                         ll_put_grouplock(llss->inode1, file1, gid);
2107                         GOTO(free, rc);
2108                 }
2109         }
2110
2111         /* to be able to restore mtime and atime after swap
2112          * we need to first save them */
2113         if (lsl->sl_flags &
2114             (SWAP_LAYOUTS_KEEP_MTIME | SWAP_LAYOUTS_KEEP_ATIME)) {
2115                 llss->ia1.ia_mtime = llss->inode1->i_mtime;
2116                 llss->ia1.ia_atime = llss->inode1->i_atime;
2117                 llss->ia1.ia_valid = ATTR_MTIME | ATTR_ATIME;
2118                 llss->ia2.ia_mtime = llss->inode2->i_mtime;
2119                 llss->ia2.ia_atime = llss->inode2->i_atime;
2120                 llss->ia2.ia_valid = ATTR_MTIME | ATTR_ATIME;
2121         }
2122
2123         /* ultimate check, before swaping the layouts we check if
2124          * dataversion has changed (if requested) */
2125         if (llss->check_dv1) {
2126                 rc = ll_data_version(llss->inode1, &dv, 0);
2127                 if (rc)
2128                         GOTO(putgl, rc);
2129                 if (dv != llss->dv1)
2130                         GOTO(putgl, rc = -EAGAIN);
2131         }
2132
2133         if (llss->check_dv2) {
2134                 rc = ll_data_version(llss->inode2, &dv, 0);
2135                 if (rc)
2136                         GOTO(putgl, rc);
2137                 if (dv != llss->dv2)
2138                         GOTO(putgl, rc = -EAGAIN);
2139         }
2140
2141         /* struct md_op_data is used to send the swap args to the mdt
2142          * only flags is missing, so we use struct mdc_swap_layouts
2143          * through the md_op_data->op_data */
2144         /* flags from user space have to be converted before they are send to
2145          * server, no flag is sent today, they are only used on the client */
2146         msl.msl_flags = 0;
2147         rc = -ENOMEM;
2148         op_data = ll_prep_md_op_data(NULL, llss->inode1, llss->inode2, NULL, 0,
2149                                      0, LUSTRE_OPC_ANY, &msl);
2150         if (IS_ERR(op_data))
2151                 GOTO(free, rc = PTR_ERR(op_data));
2152
2153         rc = obd_iocontrol(LL_IOC_LOV_SWAP_LAYOUTS, ll_i2mdexp(llss->inode1),
2154                            sizeof(*op_data), op_data, NULL);
2155         ll_finish_md_op_data(op_data);
2156
2157 putgl:
2158         if (gid != 0) {
2159                 ll_put_grouplock(llss->inode2, file2, gid);
2160                 ll_put_grouplock(llss->inode1, file1, gid);
2161         }
2162
2163         /* rc can be set from obd_iocontrol() or from a GOTO(putgl, ...) */
2164         if (rc != 0)
2165                 GOTO(free, rc);
2166
2167         /* clear useless flags */
2168         if (!(lsl->sl_flags & SWAP_LAYOUTS_KEEP_MTIME)) {
2169                 llss->ia1.ia_valid &= ~ATTR_MTIME;
2170                 llss->ia2.ia_valid &= ~ATTR_MTIME;
2171         }
2172
2173         if (!(lsl->sl_flags & SWAP_LAYOUTS_KEEP_ATIME)) {
2174                 llss->ia1.ia_valid &= ~ATTR_ATIME;
2175                 llss->ia2.ia_valid &= ~ATTR_ATIME;
2176         }
2177
2178         /* update time if requested */
2179         rc = 0;
2180         if (llss->ia2.ia_valid != 0) {
2181                 mutex_lock(&llss->inode1->i_mutex);
2182                 rc = ll_setattr(file1->f_dentry, &llss->ia2);
2183                 mutex_unlock(&llss->inode1->i_mutex);
2184         }
2185
2186         if (llss->ia1.ia_valid != 0) {
2187                 int rc1;
2188
2189                 mutex_lock(&llss->inode2->i_mutex);
2190                 rc1 = ll_setattr(file2->f_dentry, &llss->ia1);
2191                 mutex_unlock(&llss->inode2->i_mutex);
2192                 if (rc == 0)
2193                         rc = rc1;
2194         }
2195
2196 free:
2197         if (llss != NULL)
2198                 OBD_FREE_PTR(llss);
2199
2200         RETURN(rc);
2201 }
2202
2203 long ll_file_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
2204 {
2205         struct inode            *inode = file->f_dentry->d_inode;
2206         struct ll_file_data     *fd = LUSTRE_FPRIVATE(file);
2207         int                      flags, rc;
2208         ENTRY;
2209
2210         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p),cmd=%x\n", inode->i_ino,
2211                inode->i_generation, inode, cmd);
2212         ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_IOCTL, 1);
2213
2214         /* asm-ppc{,64} declares TCGETS, et. al. as type 't' not 'T' */
2215         if (_IOC_TYPE(cmd) == 'T' || _IOC_TYPE(cmd) == 't') /* tty ioctls */
2216                 RETURN(-ENOTTY);
2217
2218         switch(cmd) {
2219         case LL_IOC_GETFLAGS:
2220                 /* Get the current value of the file flags */
2221                 return put_user(fd->fd_flags, (int *)arg);
2222         case LL_IOC_SETFLAGS:
2223         case LL_IOC_CLRFLAGS:
2224                 /* Set or clear specific file flags */
2225                 /* XXX This probably needs checks to ensure the flags are
2226                  *     not abused, and to handle any flag side effects.
2227                  */
2228                 if (get_user(flags, (int *) arg))
2229                         RETURN(-EFAULT);
2230
2231                 if (cmd == LL_IOC_SETFLAGS) {
2232                         if ((flags & LL_FILE_IGNORE_LOCK) &&
2233                             !(file->f_flags & O_DIRECT)) {
2234                                 CERROR("%s: unable to disable locking on "
2235                                        "non-O_DIRECT file\n", current->comm);
2236                                 RETURN(-EINVAL);
2237                         }
2238
2239                         fd->fd_flags |= flags;
2240                 } else {
2241                         fd->fd_flags &= ~flags;
2242                 }
2243                 RETURN(0);
2244         case LL_IOC_LOV_SETSTRIPE:
2245                 RETURN(ll_lov_setstripe(inode, file, arg));
2246         case LL_IOC_LOV_SETEA:
2247                 RETURN(ll_lov_setea(inode, file, arg));
2248         case LL_IOC_LOV_SWAP_LAYOUTS: {
2249                 struct file *file2;
2250                 struct lustre_swap_layouts lsl;
2251
2252                 if (copy_from_user(&lsl, (char *)arg,
2253                                        sizeof(struct lustre_swap_layouts)))
2254                         RETURN(-EFAULT);
2255
2256                 if ((file->f_flags & O_ACCMODE) == 0) /* O_RDONLY */
2257                         RETURN(-EPERM);
2258
2259                 file2 = fget(lsl.sl_fd);
2260                 if (file2 == NULL)
2261                         RETURN(-EBADF);
2262
2263                 rc = -EPERM;
2264                 if ((file2->f_flags & O_ACCMODE) != 0) /* O_WRONLY or O_RDWR */
2265                         rc = ll_swap_layouts(file, file2, &lsl);
2266                 fput(file2);
2267                 RETURN(rc);
2268         }
2269         case LL_IOC_LOV_GETSTRIPE:
2270                 RETURN(ll_lov_getstripe(inode, arg));
2271         case LL_IOC_RECREATE_OBJ:
2272                 RETURN(ll_lov_recreate_obj(inode, arg));
2273         case LL_IOC_RECREATE_FID:
2274                 RETURN(ll_lov_recreate_fid(inode, arg));
2275         case FSFILT_IOC_FIEMAP:
2276                 RETURN(ll_ioctl_fiemap(inode, arg));
2277         case FSFILT_IOC_GETFLAGS:
2278         case FSFILT_IOC_SETFLAGS:
2279                 RETURN(ll_iocontrol(inode, file, cmd, arg));
2280         case FSFILT_IOC_GETVERSION_OLD:
2281         case FSFILT_IOC_GETVERSION:
2282                 RETURN(put_user(inode->i_generation, (int *)arg));
2283         case LL_IOC_GROUP_LOCK:
2284                 RETURN(ll_get_grouplock(inode, file, arg));
2285         case LL_IOC_GROUP_UNLOCK:
2286                 RETURN(ll_put_grouplock(inode, file, arg));
2287         case IOC_OBD_STATFS:
2288                 RETURN(ll_obd_statfs(inode, (void *)arg));
2289
2290         /* We need to special case any other ioctls we want to handle,
2291          * to send them to the MDS/OST as appropriate and to properly
2292          * network encode the arg field.
2293         case FSFILT_IOC_SETVERSION_OLD:
2294         case FSFILT_IOC_SETVERSION:
2295         */
2296         case LL_IOC_FLUSHCTX:
2297                 RETURN(ll_flush_ctx(inode));
2298         case LL_IOC_PATH2FID: {
2299                 if (copy_to_user((void *)arg, ll_inode2fid(inode),
2300                                  sizeof(struct lu_fid)))
2301                         RETURN(-EFAULT);
2302
2303                 RETURN(0);
2304         }
2305         case OBD_IOC_FID2PATH:
2306                 RETURN(ll_fid2path(inode, (void *)arg));
2307         case LL_IOC_DATA_VERSION: {
2308                 struct ioc_data_version idv;
2309                 int                     rc;
2310
2311                 if (copy_from_user(&idv, (char *)arg, sizeof(idv)))
2312                         RETURN(-EFAULT);
2313
2314                 rc = ll_data_version(inode, &idv.idv_version,
2315                                 !(idv.idv_flags & LL_DV_NOFLUSH));
2316
2317                 if (rc == 0 && copy_to_user((char *) arg, &idv, sizeof(idv)))
2318                         RETURN(-EFAULT);
2319
2320                 RETURN(rc);
2321         }
2322
2323         case LL_IOC_GET_MDTIDX: {
2324                 int mdtidx;
2325
2326                 mdtidx = ll_get_mdt_idx(inode);
2327                 if (mdtidx < 0)
2328                         RETURN(mdtidx);
2329
2330                 if (put_user((int)mdtidx, (int*)arg))
2331                         RETURN(-EFAULT);
2332
2333                 RETURN(0);
2334         }
2335         case OBD_IOC_GETDTNAME:
2336         case OBD_IOC_GETMDNAME:
2337                 RETURN(ll_get_obd_name(inode, cmd, arg));
2338         case LL_IOC_HSM_STATE_GET: {
2339                 struct md_op_data       *op_data;
2340                 struct hsm_user_state   *hus;
2341                 int                      rc;
2342
2343                 OBD_ALLOC_PTR(hus);
2344                 if (hus == NULL)
2345                         RETURN(-ENOMEM);
2346
2347                 op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL, 0, 0,
2348                                              LUSTRE_OPC_ANY, hus);
2349                 if (IS_ERR(op_data)) {
2350                         OBD_FREE_PTR(hus);
2351                         RETURN(PTR_ERR(op_data));
2352                 }
2353
2354                 rc = obd_iocontrol(cmd, ll_i2mdexp(inode), sizeof(*op_data),
2355                                    op_data, NULL);
2356
2357                 if (copy_to_user((void *)arg, hus, sizeof(*hus)))
2358                         rc = -EFAULT;
2359
2360                 ll_finish_md_op_data(op_data);
2361                 OBD_FREE_PTR(hus);
2362                 RETURN(rc);
2363         }
2364         case LL_IOC_HSM_STATE_SET: {
2365                 struct md_op_data       *op_data;
2366                 struct hsm_state_set    *hss;
2367                 int                      rc;
2368
2369                 OBD_ALLOC_PTR(hss);
2370                 if (hss == NULL)
2371                         RETURN(-ENOMEM);
2372                 if (copy_from_user(hss, (char *)arg, sizeof(*hss))) {
2373                         OBD_FREE_PTR(hss);
2374                         RETURN(-EFAULT);
2375                 }
2376
2377                 /* Non-root users are forbidden to set or clear flags which are
2378                  * NOT defined in HSM_USER_MASK. */
2379                 if (((hss->hss_setmask | hss->hss_clearmask) & ~HSM_USER_MASK)
2380                     && !cfs_capable(CFS_CAP_SYS_ADMIN)) {
2381                         OBD_FREE_PTR(hss);
2382                         RETURN(-EPERM);
2383                 }
2384
2385                 op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL, 0, 0,
2386                                              LUSTRE_OPC_ANY, hss);
2387                 if (IS_ERR(op_data)) {
2388                         OBD_FREE_PTR(hss);
2389                         RETURN(PTR_ERR(op_data));
2390                 }
2391
2392                 rc = obd_iocontrol(cmd, ll_i2mdexp(inode), sizeof(*op_data),
2393                                    op_data, NULL);
2394
2395                 ll_finish_md_op_data(op_data);
2396
2397                 OBD_FREE_PTR(hss);
2398                 RETURN(rc);
2399         }
2400         case LL_IOC_HSM_ACTION: {
2401                 struct md_op_data               *op_data;
2402                 struct hsm_current_action       *hca;
2403                 int                              rc;
2404
2405                 OBD_ALLOC_PTR(hca);
2406                 if (hca == NULL)
2407                         RETURN(-ENOMEM);
2408
2409                 op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL, 0, 0,
2410                                              LUSTRE_OPC_ANY, hca);
2411                 if (IS_ERR(op_data)) {
2412                         OBD_FREE_PTR(hca);
2413                         RETURN(PTR_ERR(op_data));
2414                 }
2415
2416                 rc = obd_iocontrol(cmd, ll_i2mdexp(inode), sizeof(*op_data),
2417                                    op_data, NULL);
2418
2419                 if (copy_to_user((char *)arg, hca, sizeof(*hca)))
2420                         rc = -EFAULT;
2421
2422                 ll_finish_md_op_data(op_data);
2423                 OBD_FREE_PTR(hca);
2424                 RETURN(rc);
2425         }
2426         case LL_IOC_SET_LEASE: {
2427                 struct ll_inode_info *lli = ll_i2info(inode);
2428                 struct obd_client_handle *och = NULL;
2429                 bool lease_broken;
2430                 fmode_t mode = 0;
2431
2432                 switch (arg) {
2433                 case F_WRLCK:
2434                         if (!(file->f_mode & FMODE_WRITE))
2435                                 RETURN(-EPERM);
2436                         mode = FMODE_WRITE;
2437                         break;
2438                 case F_RDLCK:
2439                         if (!(file->f_mode & FMODE_READ))
2440                                 RETURN(-EPERM);
2441                         mode = FMODE_READ;
2442                         break;
2443                 case F_UNLCK:
2444                         mutex_lock(&lli->lli_och_mutex);
2445                         if (fd->fd_lease_och != NULL) {
2446                                 och = fd->fd_lease_och;
2447                                 fd->fd_lease_och = NULL;
2448                         }
2449                         mutex_unlock(&lli->lli_och_mutex);
2450
2451                         if (och != NULL) {
2452                                 mode = och->och_flags &(FMODE_READ|FMODE_WRITE);
2453                                 rc = ll_lease_close(och, inode, &lease_broken);
2454                                 if (rc == 0 && lease_broken)
2455                                         mode = 0;
2456                         } else {
2457                                 rc = -ENOLCK;
2458                         }
2459
2460                         /* return the type of lease or error */
2461                         RETURN(rc < 0 ? rc : (int)mode);
2462                 default:
2463                         RETURN(-EINVAL);
2464                 }
2465
2466                 CDEBUG(D_INODE, "Set lease with mode %d\n", mode);
2467
2468                 /* apply for lease */
2469                 och = ll_lease_open(inode, file, mode);
2470                 if (IS_ERR(och))
2471                         RETURN(PTR_ERR(och));
2472
2473                 rc = 0;
2474                 mutex_lock(&lli->lli_och_mutex);
2475                 if (fd->fd_lease_och == NULL) {
2476                         fd->fd_lease_och = och;
2477                         och = NULL;
2478                 }
2479                 mutex_unlock(&lli->lli_och_mutex);
2480                 if (och != NULL) {
2481                         /* impossible now that only excl is supported for now */
2482                         ll_lease_close(och, inode, &lease_broken);
2483                         rc = -EBUSY;
2484                 }
2485                 RETURN(rc);
2486         }
2487         case LL_IOC_GET_LEASE: {
2488                 struct ll_inode_info *lli = ll_i2info(inode);
2489                 struct ldlm_lock *lock = NULL;
2490
2491                 rc = 0;
2492                 mutex_lock(&lli->lli_och_mutex);
2493                 if (fd->fd_lease_och != NULL) {
2494                         struct obd_client_handle *och = fd->fd_lease_och;
2495
2496                         lock = ldlm_handle2lock(&och->och_lease_handle);
2497                         if (lock != NULL) {
2498                                 lock_res_and_lock(lock);
2499                                 if (!ldlm_is_cancel(lock))
2500                                         rc = och->och_flags &
2501                                                 (FMODE_READ | FMODE_WRITE);
2502                                 unlock_res_and_lock(lock);
2503                                 ldlm_lock_put(lock);
2504                         }
2505                 }
2506                 mutex_unlock(&lli->lli_och_mutex);
2507
2508                 RETURN(rc);
2509         }
2510         default: {
2511                 int err;
2512
2513                 if (LLIOC_STOP ==
2514                      ll_iocontrol_call(inode, file, cmd, arg, &err))
2515                         RETURN(err);
2516
2517                 RETURN(obd_iocontrol(cmd, ll_i2dtexp(inode), 0, NULL,
2518                                      (void *)arg));
2519         }
2520         }
2521 }
2522
2523 #ifndef HAVE_FILE_LLSEEK_SIZE
2524 static inline loff_t
2525 llseek_execute(struct file *file, loff_t offset, loff_t maxsize)
2526 {
2527         if (offset < 0 && !(file->f_mode & FMODE_UNSIGNED_OFFSET))
2528                 return -EINVAL;
2529         if (offset > maxsize)
2530                 return -EINVAL;
2531
2532         if (offset != file->f_pos) {
2533                 file->f_pos = offset;
2534                 file->f_version = 0;
2535         }
2536         return offset;
2537 }
2538
2539 static loff_t
2540 generic_file_llseek_size(struct file *file, loff_t offset, int origin,
2541                 loff_t maxsize, loff_t eof)
2542 {
2543         struct inode *inode = file->f_dentry->d_inode;
2544
2545         switch (origin) {
2546         case SEEK_END:
2547                 offset += eof;
2548                 break;
2549         case SEEK_CUR:
2550                 /*
2551                  * Here we special-case the lseek(fd, 0, SEEK_CUR)
2552                  * position-querying operation.  Avoid rewriting the "same"
2553                  * f_pos value back to the file because a concurrent read(),
2554                  * write() or lseek() might have altered it
2555                  */
2556                 if (offset == 0)
2557                         return file->f_pos;
2558                 /*
2559                  * f_lock protects against read/modify/write race with other
2560                  * SEEK_CURs. Note that parallel writes and reads behave
2561                  * like SEEK_SET.
2562                  */
2563                 mutex_lock(&inode->i_mutex);
2564                 offset = llseek_execute(file, file->f_pos + offset, maxsize);
2565                 mutex_unlock(&inode->i_mutex);
2566                 return offset;
2567         case SEEK_DATA:
2568                 /*
2569                  * In the generic case the entire file is data, so as long as
2570                  * offset isn't at the end of the file then the offset is data.
2571                  */
2572                 if (offset >= eof)
2573                         return -ENXIO;
2574                 break;
2575         case SEEK_HOLE:
2576                 /*
2577                  * There is a virtual hole at the end of the file, so as long as
2578                  * offset isn't i_size or larger, return i_size.
2579                  */
2580                 if (offset >= eof)
2581                         return -ENXIO;
2582                 offset = eof;
2583                 break;
2584         }
2585
2586         return llseek_execute(file, offset, maxsize);
2587 }
2588 #endif
2589
2590 loff_t ll_file_seek(struct file *file, loff_t offset, int origin)
2591 {
2592         struct inode *inode = file->f_dentry->d_inode;
2593         loff_t retval, eof = 0;
2594
2595         ENTRY;
2596         retval = offset + ((origin == SEEK_END) ? i_size_read(inode) :
2597                            (origin == SEEK_CUR) ? file->f_pos : 0);
2598         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p), to=%llu=%#llx(%d)\n",
2599                inode->i_ino, inode->i_generation, inode, retval, retval,
2600                origin);
2601         ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_LLSEEK, 1);
2602
2603         if (origin == SEEK_END || origin == SEEK_HOLE || origin == SEEK_DATA) {
2604                 retval = ll_glimpse_size(inode);
2605                 if (retval != 0)
2606                         RETURN(retval);
2607                 eof = i_size_read(inode);
2608         }
2609
2610         retval = ll_generic_file_llseek_size(file, offset, origin,
2611                                           ll_file_maxbytes(inode), eof);
2612         RETURN(retval);
2613 }
2614
2615 int ll_flush(struct file *file, fl_owner_t id)
2616 {
2617         struct inode *inode = file->f_dentry->d_inode;
2618         struct ll_inode_info *lli = ll_i2info(inode);
2619         struct ll_file_data *fd = LUSTRE_FPRIVATE(file);
2620         int rc, err;
2621
2622         LASSERT(!S_ISDIR(inode->i_mode));
2623
2624         /* catch async errors that were recorded back when async writeback
2625          * failed for pages in this mapping. */
2626         rc = lli->lli_async_rc;
2627         lli->lli_async_rc = 0;
2628         err = lov_read_and_clear_async_rc(lli->lli_clob);
2629         if (rc == 0)
2630                 rc = err;
2631
2632         /* The application has been told write failure already.
2633          * Do not report failure again. */
2634         if (fd->fd_write_failed)
2635                 return 0;
2636         return rc ? -EIO : 0;
2637 }
2638
2639 /**
2640  * Called to make sure a portion of file has been written out.
2641  * if @local_only is not true, it will send OST_SYNC RPCs to ost.
2642  *
2643  * Return how many pages have been written.
2644  */
2645 int cl_sync_file_range(struct inode *inode, loff_t start, loff_t end,
2646                        enum cl_fsync_mode mode, int ignore_layout)
2647 {
2648         struct cl_env_nest nest;
2649         struct lu_env *env;
2650         struct cl_io *io;
2651         struct obd_capa *capa = NULL;
2652         struct cl_fsync_io *fio;
2653         int result;
2654         ENTRY;
2655
2656         if (mode != CL_FSYNC_NONE && mode != CL_FSYNC_LOCAL &&
2657             mode != CL_FSYNC_DISCARD && mode != CL_FSYNC_ALL)
2658                 RETURN(-EINVAL);
2659
2660         env = cl_env_nested_get(&nest);
2661         if (IS_ERR(env))
2662                 RETURN(PTR_ERR(env));
2663
2664         capa = ll_osscapa_get(inode, CAPA_OPC_OSS_WRITE);
2665
2666         io = ccc_env_thread_io(env);
2667         io->ci_obj = cl_i2info(inode)->lli_clob;
2668         io->ci_ignore_layout = ignore_layout;
2669
2670         /* initialize parameters for sync */
2671         fio = &io->u.ci_fsync;
2672         fio->fi_capa = capa;
2673         fio->fi_start = start;
2674         fio->fi_end = end;
2675         fio->fi_fid = ll_inode2fid(inode);
2676         fio->fi_mode = mode;
2677         fio->fi_nr_written = 0;
2678
2679         if (cl_io_init(env, io, CIT_FSYNC, io->ci_obj) == 0)
2680                 result = cl_io_loop(env, io);
2681         else
2682                 result = io->ci_result;
2683         if (result == 0)
2684                 result = fio->fi_nr_written;
2685         cl_io_fini(env, io);
2686         cl_env_nested_put(&nest, env);
2687
2688         capa_put(capa);
2689
2690         RETURN(result);
2691 }
2692
2693 /*
2694  * When dentry is provided (the 'else' case), *file->f_dentry may be
2695  * null and dentry must be used directly rather than pulled from
2696  * *file->f_dentry as is done otherwise.
2697  */
2698
2699 #ifdef HAVE_FILE_FSYNC_4ARGS
2700 int ll_fsync(struct file *file, loff_t start, loff_t end, int datasync)
2701 {
2702         struct dentry *dentry = file->f_dentry;
2703 #elif defined(HAVE_FILE_FSYNC_2ARGS)
2704 int ll_fsync(struct file *file, int datasync)
2705 {
2706         struct dentry *dentry = file->f_dentry;
2707 #else
2708 int ll_fsync(struct file *file, struct dentry *dentry, int datasync)
2709 {
2710 #endif
2711         struct inode *inode = dentry->d_inode;
2712         struct ll_inode_info *lli = ll_i2info(inode);
2713         struct ptlrpc_request *req;
2714         struct obd_capa *oc;
2715         int rc, err;
2716         ENTRY;
2717
2718         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p)\n", inode->i_ino,
2719                inode->i_generation, inode);
2720         ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_FSYNC, 1);
2721
2722 #ifdef HAVE_FILE_FSYNC_4ARGS
2723         rc = filemap_write_and_wait_range(inode->i_mapping, start, end);
2724         mutex_lock(&inode->i_mutex);
2725 #else
2726         /* fsync's caller has already called _fdata{sync,write}, we want
2727          * that IO to finish before calling the osc and mdc sync methods */
2728         rc = filemap_fdatawait(inode->i_mapping);
2729 #endif
2730
2731         /* catch async errors that were recorded back when async writeback
2732          * failed for pages in this mapping. */
2733         if (!S_ISDIR(inode->i_mode)) {
2734                 err = lli->lli_async_rc;
2735                 lli->lli_async_rc = 0;
2736                 if (rc == 0)
2737                         rc = err;
2738                 err = lov_read_and_clear_async_rc(lli->lli_clob);
2739                 if (rc == 0)
2740                         rc = err;
2741         }
2742
2743         oc = ll_mdscapa_get(inode);
2744         err = md_sync(ll_i2sbi(inode)->ll_md_exp, ll_inode2fid(inode), oc,
2745                       &req);
2746         capa_put(oc);
2747         if (!rc)
2748                 rc = err;
2749         if (!err)
2750                 ptlrpc_req_finished(req);
2751
2752         if (datasync && S_ISREG(inode->i_mode)) {
2753                 struct ll_file_data *fd = LUSTRE_FPRIVATE(file);
2754
2755                 err = cl_sync_file_range(inode, 0, OBD_OBJECT_EOF,
2756                                 CL_FSYNC_ALL, 0);
2757                 if (rc == 0 && err < 0)
2758                         rc = err;
2759                 if (rc < 0)
2760                         fd->fd_write_failed = true;
2761                 else
2762                         fd->fd_write_failed = false;
2763         }
2764
2765 #ifdef HAVE_FILE_FSYNC_4ARGS
2766         mutex_unlock(&inode->i_mutex);
2767 #endif
2768         RETURN(rc);
2769 }
2770
2771 int ll_file_flock(struct file *file, int cmd, struct file_lock *file_lock)
2772 {
2773         struct inode *inode = file->f_dentry->d_inode;
2774         struct ll_sb_info *sbi = ll_i2sbi(inode);
2775         struct ldlm_enqueue_info einfo = {
2776                 .ei_type        = LDLM_FLOCK,
2777                 .ei_cb_cp       = ldlm_flock_completion_ast,
2778                 .ei_cbdata      = file_lock,
2779         };
2780         struct md_op_data *op_data;
2781         struct lustre_handle lockh = {0};
2782         ldlm_policy_data_t flock = {{0}};
2783         int flags = 0;
2784         int rc;
2785         int rc2 = 0;
2786         ENTRY;
2787
2788         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu file_lock=%p\n",
2789                inode->i_ino, file_lock);
2790
2791         ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_FLOCK, 1);
2792
2793         if (file_lock->fl_flags & FL_FLOCK) {
2794                 LASSERT((cmd == F_SETLKW) || (cmd == F_SETLK));
2795                 /* flocks are whole-file locks */
2796                 flock.l_flock.end = OFFSET_MAX;
2797                 /* For flocks owner is determined by the local file desctiptor*/
2798                 flock.l_flock.owner = (unsigned long)file_lock->fl_file;
2799         } else if (file_lock->fl_flags & FL_POSIX) {
2800                 flock.l_flock.owner = (unsigned long)file_lock->fl_owner;
2801                 flock.l_flock.start = file_lock->fl_start;
2802                 flock.l_flock.end = file_lock->fl_end;
2803         } else {
2804                 RETURN(-EINVAL);
2805         }
2806         flock.l_flock.pid = file_lock->fl_pid;
2807
2808         /* Somewhat ugly workaround for svc lockd.
2809          * lockd installs custom fl_lmops->lm_compare_owner that checks
2810          * for the fl_owner to be the same (which it always is on local node
2811          * I guess between lockd processes) and then compares pid.
2812          * As such we assign pid to the owner field to make it all work,
2813          * conflict with normal locks is unlikely since pid space and
2814          * pointer space for current->files are not intersecting */
2815         if (file_lock->fl_lmops && file_lock->fl_lmops->lm_compare_owner)
2816                 flock.l_flock.owner = (unsigned long)file_lock->fl_pid;
2817
2818         switch (file_lock->fl_type) {
2819         case F_RDLCK:
2820                 einfo.ei_mode = LCK_PR;
2821                 break;
2822         case F_UNLCK:
2823                 /* An unlock request may or may not have any relation to
2824                  * existing locks so we may not be able to pass a lock handle
2825                  * via a normal ldlm_lock_cancel() request. The request may even
2826                  * unlock a byte range in the middle of an existing lock. In
2827                  * order to process an unlock request we need all of the same
2828                  * information that is given with a normal read or write record
2829                  * lock request. To avoid creating another ldlm unlock (cancel)
2830                  * message we'll treat a LCK_NL flock request as an unlock. */
2831                 einfo.ei_mode = LCK_NL;
2832                 break;
2833         case F_WRLCK:
2834                 einfo.ei_mode = LCK_PW;
2835                 break;
2836         default:
2837                 CDEBUG(D_INFO, "Unknown fcntl lock type: %d\n",
2838                         file_lock->fl_type);
2839                 RETURN (-ENOTSUPP);
2840         }
2841
2842         switch (cmd) {
2843         case F_SETLKW:
2844 #ifdef F_SETLKW64
2845         case F_SETLKW64:
2846 #endif
2847                 flags = 0;
2848                 break;
2849         case F_SETLK:
2850 #ifdef F_SETLK64
2851         case F_SETLK64:
2852 #endif
2853                 flags = LDLM_FL_BLOCK_NOWAIT;
2854                 break;
2855         case F_GETLK:
2856 #ifdef F_GETLK64
2857         case F_GETLK64:
2858 #endif
2859                 flags = LDLM_FL_TEST_LOCK;
2860                 /* Save the old mode so that if the mode in the lock changes we
2861                  * can decrement the appropriate reader or writer refcount. */
2862                 file_lock->fl_type = einfo.ei_mode;
2863                 break;
2864         default:
2865                 CERROR("unknown fcntl lock command: %d\n", cmd);
2866                 RETURN (-EINVAL);
2867         }
2868
2869         op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL, 0, 0,
2870                                      LUSTRE_OPC_ANY, NULL);
2871         if (IS_ERR(op_data))
2872                 RETURN(PTR_ERR(op_data));
2873
2874         CDEBUG(D_DLMTRACE, "inode=%lu, pid=%u, flags=%#x, mode=%u, "
2875                "start="LPU64", end="LPU64"\n", inode->i_ino, flock.l_flock.pid,
2876                flags, einfo.ei_mode, flock.l_flock.start, flock.l_flock.end);
2877
2878         rc = md_enqueue(sbi->ll_md_exp, &einfo, NULL,
2879                         op_data, &lockh, &flock, 0, NULL /* req */, flags);
2880
2881         if ((file_lock->fl_flags & FL_FLOCK) &&
2882             (rc == 0 || file_lock->fl_type == F_UNLCK))
2883                 rc2  = flock_lock_file_wait(file, file_lock);
2884         if ((file_lock->fl_flags & FL_POSIX) &&
2885             (rc == 0 || file_lock->fl_type == F_UNLCK) &&
2886             !(flags & LDLM_FL_TEST_LOCK))
2887                 rc2  = posix_lock_file_wait(file, file_lock);
2888
2889         if (rc2 && file_lock->fl_type != F_UNLCK) {
2890                 einfo.ei_mode = LCK_NL;
2891                 md_enqueue(sbi->ll_md_exp, &einfo, NULL,
2892                         op_data, &lockh, &flock, 0, NULL /* req */, flags);
2893                 rc = rc2;
2894         }
2895
2896         ll_finish_md_op_data(op_data);
2897
2898         RETURN(rc);
2899 }
2900
2901 int ll_file_noflock(struct file *file, int cmd, struct file_lock *file_lock)
2902 {
2903         ENTRY;
2904
2905         RETURN(-ENOSYS);
2906 }
2907
2908 /**
2909  * test if some locks matching bits and l_req_mode are acquired
2910  * - bits can be in different locks
2911  * - if found clear the common lock bits in *bits
2912  * - the bits not found, are kept in *bits
2913  * \param inode [IN]
2914  * \param bits [IN] searched lock bits [IN]
2915  * \param l_req_mode [IN] searched lock mode
2916  * \retval boolean, true iff all bits are found
2917  */
2918 int ll_have_md_lock(struct inode *inode, __u64 *bits,  ldlm_mode_t l_req_mode)
2919 {
2920         struct lustre_handle lockh;
2921         ldlm_policy_data_t policy;
2922         ldlm_mode_t mode = (l_req_mode == LCK_MINMODE) ?
2923                                 (LCK_CR|LCK_CW|LCK_PR|LCK_PW) : l_req_mode;
2924         struct lu_fid *fid;
2925         __u64 flags;
2926         int i;
2927         ENTRY;
2928
2929         if (!inode)
2930                RETURN(0);
2931
2932         fid = &ll_i2info(inode)->lli_fid;
2933         CDEBUG(D_INFO, "trying to match res "DFID" mode %s\n", PFID(fid),
2934                ldlm_lockname[mode]);
2935
2936         flags = LDLM_FL_BLOCK_GRANTED | LDLM_FL_CBPENDING | LDLM_FL_TEST_LOCK;
2937         for (i = 0; i <= MDS_INODELOCK_MAXSHIFT && *bits != 0; i++) {
2938                 policy.l_inodebits.bits = *bits & (1 << i);
2939                 if (policy.l_inodebits.bits == 0)
2940                         continue;
2941
2942                 if (md_lock_match(ll_i2mdexp(inode), flags, fid, LDLM_IBITS,
2943                                   &policy, mode, &lockh)) {
2944                         struct ldlm_lock *lock;
2945
2946                         lock = ldlm_handle2lock(&lockh);
2947                         if (lock) {
2948                                 *bits &=
2949                                       ~(lock->l_policy_data.l_inodebits.bits);
2950                                 LDLM_LOCK_PUT(lock);
2951                         } else {
2952                                 *bits &= ~policy.l_inodebits.bits;
2953                         }
2954                 }
2955         }
2956         RETURN(*bits == 0);
2957 }
2958
2959 ldlm_mode_t ll_take_md_lock(struct inode *inode, __u64 bits,
2960                             struct lustre_handle *lockh, __u64 flags)
2961 {
2962         ldlm_policy_data_t policy = { .l_inodebits = {bits}};
2963         struct lu_fid *fid;
2964         ldlm_mode_t rc;
2965         ENTRY;
2966
2967         fid = &ll_i2info(inode)->lli_fid;
2968         CDEBUG(D_INFO, "trying to match res "DFID"\n", PFID(fid));
2969
2970         rc = md_lock_match(ll_i2mdexp(inode), LDLM_FL_BLOCK_GRANTED|flags,
2971                            fid, LDLM_IBITS, &policy,
2972                            LCK_CR|LCK_CW|LCK_PR|LCK_PW, lockh);
2973         RETURN(rc);
2974 }
2975
2976 static int ll_inode_revalidate_fini(struct inode *inode, int rc)
2977 {
2978         /* Already unlinked. Just update nlink and return success */
2979         if (rc == -ENOENT) {
2980                 clear_nlink(inode);
2981                 /* This path cannot be hit for regular files unless in
2982                  * case of obscure races, so no need to to validate
2983                  * size. */
2984                 if (!S_ISREG(inode->i_mode) && !S_ISDIR(inode->i_mode))
2985                         return 0;
2986         } else if (rc != 0) {
2987                 CERROR("%s: revalidate FID "DFID" error: rc = %d\n",
2988                        ll_get_fsname(inode->i_sb, NULL, 0),
2989                        PFID(ll_inode2fid(inode)), rc);
2990         }
2991
2992         return rc;
2993 }
2994
2995 int __ll_inode_revalidate_it(struct dentry *dentry, struct lookup_intent *it,
2996                              __u64 ibits)
2997 {
2998         struct inode *inode = dentry->d_inode;
2999         struct ptlrpc_request *req = NULL;
3000         struct obd_export *exp;
3001         int rc = 0;
3002         ENTRY;
3003
3004         LASSERT(inode != NULL);
3005
3006         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p),name=%s\n",
3007                inode->i_ino, inode->i_generation, inode, dentry->d_name.name);
3008
3009         exp = ll_i2mdexp(inode);
3010
3011         /* XXX: Enable OBD_CONNECT_ATTRFID to reduce unnecessary getattr RPC.
3012          *      But under CMD case, it caused some lock issues, should be fixed
3013          *      with new CMD ibits lock. See bug 12718 */
3014         if (exp_connect_flags(exp) & OBD_CONNECT_ATTRFID) {
3015                 struct lookup_intent oit = { .it_op = IT_GETATTR };
3016                 struct md_op_data *op_data;
3017
3018                 if (ibits == MDS_INODELOCK_LOOKUP)
3019                         oit.it_op = IT_LOOKUP;
3020
3021                 /* Call getattr by fid, so do not provide name at all. */
3022                 op_data = ll_prep_md_op_data(NULL, dentry->d_parent->d_inode,
3023                                              dentry->d_inode, NULL, 0, 0,
3024                                              LUSTRE_OPC_ANY, NULL);
3025                 if (IS_ERR(op_data))
3026                         RETURN(PTR_ERR(op_data));
3027
3028                 oit.it_create_mode |= M_CHECK_STALE;
3029                 rc = md_intent_lock(exp, op_data, NULL, 0,
3030                                     /* we are not interested in name
3031                                        based lookup */
3032                                     &oit, 0, &req,
3033                                     ll_md_blocking_ast, 0);
3034                 ll_finish_md_op_data(op_data);
3035                 oit.it_create_mode &= ~M_CHECK_STALE;
3036                 if (rc < 0) {
3037                         rc = ll_inode_revalidate_fini(inode, rc);
3038                         GOTO (out, rc);
3039                 }
3040
3041                 rc = ll_revalidate_it_finish(req, &oit, dentry);
3042                 if (rc != 0) {
3043                         ll_intent_release(&oit);
3044                         GOTO(out, rc);
3045                 }
3046
3047                 /* Unlinked? Unhash dentry, so it is not picked up later by
3048                    do_lookup() -> ll_revalidate_it(). We cannot use d_drop
3049                    here to preserve get_cwd functionality on 2.6.
3050                    Bug 10503 */
3051                 if (!dentry->d_inode->i_nlink)
3052                         d_lustre_invalidate(dentry, 0);
3053
3054                 ll_lookup_finish_locks(&oit, dentry);
3055         } else if (!ll_have_md_lock(dentry->d_inode, &ibits, LCK_MINMODE)) {
3056                 struct ll_sb_info *sbi = ll_i2sbi(dentry->d_inode);
3057                 obd_valid valid = OBD_MD_FLGETATTR;
3058                 struct md_op_data *op_data;
3059                 int ealen = 0;
3060
3061                 if (S_ISREG(inode->i_mode)) {
3062                         rc = ll_get_max_mdsize(sbi, &ealen);
3063                         if (rc)
3064                                 RETURN(rc);
3065                         valid |= OBD_MD_FLEASIZE | OBD_MD_FLMODEASIZE;
3066                 }
3067
3068                 op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL,
3069                                              0, ealen, LUSTRE_OPC_ANY,
3070                                              NULL);
3071                 if (IS_ERR(op_data))
3072                         RETURN(PTR_ERR(op_data));
3073
3074                 op_data->op_valid = valid;
3075                 /* Once OBD_CONNECT_ATTRFID is not supported, we can't find one
3076                  * capa for this inode. Because we only keep capas of dirs
3077                  * fresh. */
3078                 rc = md_getattr(sbi->ll_md_exp, op_data, &req);
3079                 ll_finish_md_op_data(op_data);
3080                 if (rc) {
3081                         rc = ll_inode_revalidate_fini(inode, rc);
3082                         RETURN(rc);
3083                 }
3084
3085                 rc = ll_prep_inode(&inode, req, NULL, NULL);
3086         }
3087 out:
3088         ptlrpc_req_finished(req);
3089         return rc;
3090 }
3091
3092 int ll_inode_revalidate_it(struct dentry *dentry, struct lookup_intent *it,
3093                            __u64 ibits)
3094 {
3095         struct inode    *inode = dentry->d_inode;
3096         int              rc;
3097         ENTRY;
3098
3099         rc = __ll_inode_revalidate_it(dentry, it, ibits);
3100         if (rc != 0)
3101                 RETURN(rc);
3102
3103         /* if object isn't regular file, don't validate size */
3104         if (!S_ISREG(inode->i_mode)) {
3105                 LTIME_S(inode->i_atime) = ll_i2info(inode)->lli_lvb.lvb_atime;
3106                 LTIME_S(inode->i_mtime) = ll_i2info(inode)->lli_lvb.lvb_mtime;
3107                 LTIME_S(inode->i_ctime) = ll_i2info(inode)->lli_lvb.lvb_ctime;
3108         } else {
3109                 /* In case of restore, the MDT has the right size and has
3110                  * already send it back without granting the layout lock,
3111                  * inode is up-to-date so glimpse is useless.
3112                  * Also to glimpse we need the layout, in case of a running
3113                  * restore the MDT holds the layout lock so the glimpse will
3114                  * block up to the end of restore (getattr will block)
3115                  */
3116                 if (!(ll_i2info(inode)->lli_flags & LLIF_FILE_RESTORING))
3117                         rc = ll_glimpse_size(inode);
3118         }
3119         RETURN(rc);
3120 }
3121
3122 int ll_getattr_it(struct vfsmount *mnt, struct dentry *de,
3123                   struct lookup_intent *it, struct kstat *stat)
3124 {
3125         struct inode *inode = de->d_inode;
3126         struct ll_sb_info *sbi = ll_i2sbi(inode);
3127         struct ll_inode_info *lli = ll_i2info(inode);
3128         int res = 0;
3129
3130         res = ll_inode_revalidate_it(de, it, MDS_INODELOCK_UPDATE |
3131                                              MDS_INODELOCK_LOOKUP);
3132         ll_stats_ops_tally(sbi, LPROC_LL_GETATTR, 1);
3133
3134         if (res)
3135                 return res;
3136
3137         stat->dev = inode->i_sb->s_dev;
3138         if (ll_need_32bit_api(sbi))
3139                 stat->ino = cl_fid_build_ino(&lli->lli_fid, 1);
3140         else
3141                 stat->ino = inode->i_ino;
3142         stat->mode = inode->i_mode;
3143         stat->nlink = inode->i_nlink;
3144         stat->uid = inode->i_uid;
3145         stat->gid = inode->i_gid;
3146         stat->rdev = inode->i_rdev;
3147         stat->atime = inode->i_atime;
3148         stat->mtime = inode->i_mtime;
3149         stat->ctime = inode->i_ctime;
3150         stat->blksize = 1 << inode->i_blkbits;
3151
3152         stat->size = i_size_read(inode);
3153         stat->blocks = inode->i_blocks;
3154
3155         return 0;
3156 }
3157 int ll_getattr(struct vfsmount *mnt, struct dentry *de, struct kstat *stat)
3158 {
3159         struct lookup_intent it = { .it_op = IT_GETATTR };
3160
3161         return ll_getattr_it(mnt, de, &it, stat);
3162 }
3163
3164 #ifdef HAVE_LINUX_FIEMAP_H
3165 int ll_fiemap(struct inode *inode, struct fiemap_extent_info *fieinfo,
3166                 __u64 start, __u64 len)
3167 {
3168         int rc;
3169         size_t num_bytes;
3170         struct ll_user_fiemap *fiemap;
3171         unsigned int extent_count = fieinfo->fi_extents_max;
3172
3173         num_bytes = sizeof(*fiemap) + (extent_count *
3174                                        sizeof(struct ll_fiemap_extent));
3175         OBD_ALLOC_LARGE(fiemap, num_bytes);
3176
3177         if (fiemap == NULL)
3178                 RETURN(-ENOMEM);
3179
3180         fiemap->fm_flags = fieinfo->fi_flags;
3181         fiemap->fm_extent_count = fieinfo->fi_extents_max;
3182         fiemap->fm_start = start;
3183         fiemap->fm_length = len;
3184         memcpy(&fiemap->fm_extents[0], fieinfo->fi_extents_start,
3185                sizeof(struct ll_fiemap_extent));
3186
3187         rc = ll_do_fiemap(inode, fiemap, num_bytes);
3188
3189         fieinfo->fi_flags = fiemap->fm_flags;
3190         fieinfo->fi_extents_mapped = fiemap->fm_mapped_extents;
3191         memcpy(fieinfo->fi_extents_start, &fiemap->fm_extents[0],
3192                fiemap->fm_mapped_extents * sizeof(struct ll_fiemap_extent));
3193
3194         OBD_FREE_LARGE(fiemap, num_bytes);
3195         return rc;
3196 }
3197 #endif
3198
3199 struct posix_acl * ll_get_acl(struct inode *inode, int type)
3200 {
3201         struct ll_inode_info *lli = ll_i2info(inode);
3202         struct posix_acl *acl = NULL;
3203         ENTRY;
3204
3205         spin_lock(&lli->lli_lock);
3206         /* VFS' acl_permission_check->check_acl will release the refcount */
3207         acl = posix_acl_dup(lli->lli_posix_acl);
3208         spin_unlock(&lli->lli_lock);
3209
3210         RETURN(acl);
3211 }
3212
3213 #ifndef HAVE_GENERIC_PERMISSION_2ARGS
3214 static int
3215 # ifdef HAVE_GENERIC_PERMISSION_4ARGS
3216 ll_check_acl(struct inode *inode, int mask, unsigned int flags)
3217 # else
3218 ll_check_acl(struct inode *inode, int mask)
3219 # endif
3220 {
3221 # ifdef CONFIG_FS_POSIX_ACL
3222         struct posix_acl *acl;
3223         int rc;
3224         ENTRY;
3225
3226 #  ifdef HAVE_GENERIC_PERMISSION_4ARGS
3227         if (flags & IPERM_FLAG_RCU)
3228                 return -ECHILD;
3229 #  endif
3230         acl = ll_get_acl(inode, ACL_TYPE_ACCESS);
3231
3232         if (!acl)
3233                 RETURN(-EAGAIN);
3234
3235         rc = posix_acl_permission(inode, acl, mask);
3236         posix_acl_release(acl);
3237
3238         RETURN(rc);
3239 # else /* !CONFIG_FS_POSIX_ACL */
3240         return -EAGAIN;
3241 # endif /* CONFIG_FS_POSIX_ACL */
3242 }
3243 #endif /* HAVE_GENERIC_PERMISSION_2ARGS */
3244
3245 #ifdef HAVE_GENERIC_PERMISSION_4ARGS
3246 int ll_inode_permission(struct inode *inode, int mask, unsigned int flags)
3247 #else
3248 # ifdef HAVE_INODE_PERMISION_2ARGS
3249 int ll_inode_permission(struct inode *inode, int mask)
3250 # else
3251 int ll_inode_permission(struct inode *inode, int mask, struct nameidata *nd)
3252 # endif
3253 #endif
3254 {
3255         int rc = 0;
3256         ENTRY;
3257
3258 #ifdef MAY_NOT_BLOCK
3259         if (mask & MAY_NOT_BLOCK)
3260                 return -ECHILD;
3261 #elif defined(HAVE_GENERIC_PERMISSION_4ARGS)
3262         if (flags & IPERM_FLAG_RCU)
3263                 return -ECHILD;
3264 #endif
3265
3266        /* as root inode are NOT getting validated in lookup operation,
3267         * need to do it before permission check. */
3268
3269         if (inode == inode->i_sb->s_root->d_inode) {
3270                 struct lookup_intent it = { .it_op = IT_LOOKUP };
3271
3272                 rc = __ll_inode_revalidate_it(inode->i_sb->s_root, &it,
3273                                               MDS_INODELOCK_LOOKUP);
3274                 if (rc)
3275                         RETURN(rc);
3276         }
3277
3278         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p), inode mode %x mask %o\n",
3279                inode->i_ino, inode->i_generation, inode, inode->i_mode, mask);
3280
3281         if (ll_i2sbi(inode)->ll_flags & LL_SBI_RMT_CLIENT)
3282                 return lustre_check_remote_perm(inode, mask);
3283
3284         ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_INODE_PERM, 1);
3285         rc = ll_generic_permission(inode, mask, flags, ll_check_acl);
3286
3287         RETURN(rc);
3288 }
3289
3290 #ifdef HAVE_FILE_READV
3291 #define READ_METHOD readv
3292 #define READ_FUNCTION ll_file_readv
3293 #define WRITE_METHOD writev
3294 #define WRITE_FUNCTION ll_file_writev
3295 #else
3296 #define READ_METHOD aio_read
3297 #define READ_FUNCTION ll_file_aio_read
3298 #define WRITE_METHOD aio_write
3299 #define WRITE_FUNCTION ll_file_aio_write
3300 #endif
3301
3302 /* -o localflock - only provides locally consistent flock locks */
3303 struct file_operations ll_file_operations = {
3304         .read           = ll_file_read,
3305         .READ_METHOD    = READ_FUNCTION,
3306         .write          = ll_file_write,
3307         .WRITE_METHOD   = WRITE_FUNCTION,
3308         .unlocked_ioctl = ll_file_ioctl,
3309         .open           = ll_file_open,
3310         .release        = ll_file_release,
3311         .mmap           = ll_file_mmap,
3312         .llseek         = ll_file_seek,
3313         .splice_read    = ll_file_splice_read,
3314         .fsync          = ll_fsync,
3315         .flush          = ll_flush
3316 };
3317
3318 struct file_operations ll_file_operations_flock = {
3319         .read           = ll_file_read,
3320         .READ_METHOD    = READ_FUNCTION,
3321         .write          = ll_file_write,
3322         .WRITE_METHOD   = WRITE_FUNCTION,
3323         .unlocked_ioctl = ll_file_ioctl,
3324         .open           = ll_file_open,
3325         .release        = ll_file_release,
3326         .mmap           = ll_file_mmap,
3327         .llseek         = ll_file_seek,
3328         .splice_read    = ll_file_splice_read,
3329         .fsync          = ll_fsync,
3330         .flush          = ll_flush,
3331         .flock          = ll_file_flock,
3332         .lock           = ll_file_flock
3333 };
3334
3335 /* These are for -o noflock - to return ENOSYS on flock calls */
3336 struct file_operations ll_file_operations_noflock = {
3337         .read           = ll_file_read,
3338         .READ_METHOD    = READ_FUNCTION,
3339         .write          = ll_file_write,
3340         .WRITE_METHOD   = WRITE_FUNCTION,
3341         .unlocked_ioctl = ll_file_ioctl,
3342         .open           = ll_file_open,
3343         .release        = ll_file_release,
3344         .mmap           = ll_file_mmap,
3345         .llseek         = ll_file_seek,
3346         .splice_read    = ll_file_splice_read,
3347         .fsync          = ll_fsync,
3348         .flush          = ll_flush,
3349         .flock          = ll_file_noflock,
3350         .lock           = ll_file_noflock
3351 };
3352
3353 struct inode_operations ll_file_inode_operations = {
3354         .setattr        = ll_setattr,
3355         .getattr        = ll_getattr,
3356         .permission     = ll_inode_permission,
3357         .setxattr       = ll_setxattr,
3358         .getxattr       = ll_getxattr,
3359         .listxattr      = ll_listxattr,
3360         .removexattr    = ll_removexattr,
3361 #ifdef  HAVE_LINUX_FIEMAP_H
3362         .fiemap         = ll_fiemap,
3363 #endif
3364 #ifdef HAVE_IOP_GET_ACL
3365         .get_acl        = ll_get_acl,
3366 #endif
3367 };
3368
3369 /* dynamic ioctl number support routins */
3370 static struct llioc_ctl_data {
3371         struct rw_semaphore     ioc_sem;
3372         cfs_list_t              ioc_head;
3373 } llioc = {
3374         __RWSEM_INITIALIZER(llioc.ioc_sem),
3375         CFS_LIST_HEAD_INIT(llioc.ioc_head)
3376 };
3377
3378
3379 struct llioc_data {
3380         cfs_list_t              iocd_list;
3381         unsigned int            iocd_size;
3382         llioc_callback_t        iocd_cb;
3383         unsigned int            iocd_count;
3384         unsigned int            iocd_cmd[0];
3385 };
3386
3387 void *ll_iocontrol_register(llioc_callback_t cb, int count, unsigned int *cmd)
3388 {
3389         unsigned int size;
3390         struct llioc_data *in_data = NULL;
3391         ENTRY;
3392
3393         if (cb == NULL || cmd == NULL ||
3394             count > LLIOC_MAX_CMD || count < 0)
3395                 RETURN(NULL);
3396
3397         size = sizeof(*in_data) + count * sizeof(unsigned int);
3398         OBD_ALLOC(in_data, size);
3399         if (in_data == NULL)
3400                 RETURN(NULL);
3401
3402         memset(in_data, 0, sizeof(*in_data));
3403         in_data->iocd_size = size;
3404         in_data->iocd_cb = cb;
3405         in_data->iocd_count = count;
3406         memcpy(in_data->iocd_cmd, cmd, sizeof(unsigned int) * count);
3407
3408         down_write(&llioc.ioc_sem);
3409         cfs_list_add_tail(&in_data->iocd_list, &llioc.ioc_head);
3410         up_write(&llioc.ioc_sem);
3411
3412         RETURN(in_data);
3413 }
3414
3415 void ll_iocontrol_unregister(void *magic)
3416 {
3417         struct llioc_data *tmp;
3418
3419         if (magic == NULL)
3420                 return;
3421
3422         down_write(&llioc.ioc_sem);
3423         cfs_list_for_each_entry(tmp, &llioc.ioc_head, iocd_list) {
3424                 if (tmp == magic) {
3425                         unsigned int size = tmp->iocd_size;
3426
3427                         cfs_list_del(&tmp->iocd_list);
3428                         up_write(&llioc.ioc_sem);
3429
3430                         OBD_FREE(tmp, size);
3431                         return;
3432                 }
3433         }
3434         up_write(&llioc.ioc_sem);
3435
3436         CWARN("didn't find iocontrol register block with magic: %p\n", magic);
3437 }
3438
3439 EXPORT_SYMBOL(ll_iocontrol_register);
3440 EXPORT_SYMBOL(ll_iocontrol_unregister);
3441
3442 enum llioc_iter ll_iocontrol_call(struct inode *inode, struct file *file,
3443                         unsigned int cmd, unsigned long arg, int *rcp)
3444 {
3445         enum llioc_iter ret = LLIOC_CONT;
3446         struct llioc_data *data;
3447         int rc = -EINVAL, i;
3448
3449         down_read(&llioc.ioc_sem);
3450         cfs_list_for_each_entry(data, &llioc.ioc_head, iocd_list) {
3451                 for (i = 0; i < data->iocd_count; i++) {
3452                         if (cmd != data->iocd_cmd[i])
3453                                 continue;
3454
3455                         ret = data->iocd_cb(inode, file, cmd, arg, data, &rc);
3456                         break;
3457                 }
3458
3459                 if (ret == LLIOC_STOP)
3460                         break;
3461         }
3462         up_read(&llioc.ioc_sem);
3463
3464         if (rcp)
3465                 *rcp = rc;
3466         return ret;
3467 }
3468
3469 int ll_layout_conf(struct inode *inode, const struct cl_object_conf *conf)
3470 {
3471         struct ll_inode_info *lli = ll_i2info(inode);
3472         struct cl_env_nest nest;
3473         struct lu_env *env;
3474         int result;
3475         ENTRY;
3476
3477         if (lli->lli_clob == NULL)
3478                 RETURN(0);
3479
3480         env = cl_env_nested_get(&nest);
3481         if (IS_ERR(env))
3482                 RETURN(PTR_ERR(env));
3483
3484         result = cl_conf_set(env, lli->lli_clob, conf);
3485         cl_env_nested_put(&nest, env);
3486
3487         if (conf->coc_opc == OBJECT_CONF_SET) {
3488                 struct ldlm_lock *lock = conf->coc_lock;
3489
3490                 LASSERT(lock != NULL);
3491                 LASSERT(ldlm_has_layout(lock));
3492                 if (result == 0) {
3493                         /* it can only be allowed to match after layout is
3494                          * applied to inode otherwise false layout would be
3495                          * seen. Applying layout shoud happen before dropping
3496                          * the intent lock. */
3497                         ldlm_lock_allow_match(lock);
3498                 }
3499         }
3500         RETURN(result);
3501 }
3502
3503 /* Fetch layout from MDT with getxattr request, if it's not ready yet */
3504 static int ll_layout_fetch(struct inode *inode, struct ldlm_lock *lock)
3505
3506 {
3507         struct ll_sb_info *sbi = ll_i2sbi(inode);
3508         struct obd_capa *oc;
3509         struct ptlrpc_request *req;
3510         struct mdt_body *body;
3511         void *lvbdata;
3512         void *lmm;
3513         int lmmsize;
3514         int rc;
3515         ENTRY;
3516
3517         CDEBUG(D_INODE, DFID" LVB_READY=%d l_lvb_data=%p l_lvb_len=%d\n",
3518                PFID(ll_inode2fid(inode)), !!(lock->l_flags & LDLM_FL_LVB_READY),
3519                lock->l_lvb_data, lock->l_lvb_len);
3520
3521         if ((lock->l_lvb_data != NULL) && (lock->l_flags & LDLM_FL_LVB_READY))
3522                 RETURN(0);
3523
3524         /* if layout lock was granted right away, the layout is returned
3525          * within DLM_LVB of dlm reply; otherwise if the lock was ever
3526          * blocked and then granted via completion ast, we have to fetch
3527          * layout here. Please note that we can't use the LVB buffer in
3528          * completion AST because it doesn't have a large enough buffer */
3529         oc = ll_mdscapa_get(inode);
3530         rc = ll_get_max_mdsize(sbi, &lmmsize);
3531         if (rc == 0)
3532                 rc = md_getxattr(sbi->ll_md_exp, ll_inode2fid(inode), oc,
3533                                 OBD_MD_FLXATTR, XATTR_NAME_LOV, NULL, 0,
3534                                 lmmsize, 0, &req);
3535         capa_put(oc);
3536         if (rc < 0)
3537                 RETURN(rc);
3538
3539         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
3540         if (body == NULL || body->eadatasize > lmmsize)
3541                 GOTO(out, rc = -EPROTO);
3542
3543         lmmsize = body->eadatasize;
3544         if (lmmsize == 0) /* empty layout */
3545                 GOTO(out, rc = 0);
3546
3547         lmm = req_capsule_server_sized_get(&req->rq_pill, &RMF_EADATA, lmmsize);
3548         if (lmm == NULL)
3549                 GOTO(out, rc = -EFAULT);
3550
3551         OBD_ALLOC_LARGE(lvbdata, lmmsize);
3552         if (lvbdata == NULL)
3553                 GOTO(out, rc = -ENOMEM);
3554
3555         memcpy(lvbdata, lmm, lmmsize);
3556         lock_res_and_lock(lock);
3557         if (lock->l_lvb_data != NULL)
3558                 OBD_FREE_LARGE(lock->l_lvb_data, lock->l_lvb_len);
3559
3560         lock->l_lvb_data = lvbdata;
3561         lock->l_lvb_len = lmmsize;
3562         unlock_res_and_lock(lock);
3563
3564         EXIT;
3565
3566 out:
3567         ptlrpc_req_finished(req);
3568         return rc;
3569 }
3570
3571 /**
3572  * Apply the layout to the inode. Layout lock is held and will be released
3573  * in this function.
3574  */
3575 static int ll_layout_lock_set(struct lustre_handle *lockh, ldlm_mode_t mode,
3576                                 struct inode *inode, __u32 *gen, bool reconf)
3577 {
3578         struct ll_inode_info *lli = ll_i2info(inode);
3579         struct ll_sb_info    *sbi = ll_i2sbi(inode);
3580         struct ldlm_lock *lock;
3581         struct lustre_md md = { NULL };
3582         struct cl_object_conf conf;
3583         int rc = 0;
3584         bool lvb_ready;
3585         bool wait_layout = false;
3586         ENTRY;
3587
3588         LASSERT(lustre_handle_is_used(lockh));
3589
3590         lock = ldlm_handle2lock(lockh);
3591         LASSERT(lock != NULL);
3592         LASSERT(ldlm_has_layout(lock));
3593
3594         LDLM_DEBUG(lock, "File %p/"DFID" being reconfigured: %d.\n",
3595                    inode, PFID(&lli->lli_fid), reconf);
3596
3597         /* in case this is a caching lock and reinstate with new inode */
3598         md_set_lock_data(sbi->ll_md_exp, &lockh->cookie, inode, NULL);
3599
3600         lock_res_and_lock(lock);
3601         lvb_ready = !!(lock->l_flags & LDLM_FL_LVB_READY);
3602         unlock_res_and_lock(lock);
3603         /* checking lvb_ready is racy but this is okay. The worst case is
3604          * that multi processes may configure the file on the same time. */
3605
3606         if (lvb_ready || !reconf) {
3607                 rc = -ENODATA;
3608                 if (lvb_ready) {
3609                         /* layout_gen must be valid if layout lock is not
3610                          * cancelled and stripe has already set */
3611                         *gen = lli->lli_layout_gen;
3612                         rc = 0;
3613                 }
3614                 GOTO(out, rc);
3615         }
3616
3617         rc = ll_layout_fetch(inode, lock);
3618         if (rc < 0)
3619                 GOTO(out, rc);
3620
3621         /* for layout lock, lmm is returned in lock's lvb.
3622          * lvb_data is immutable if the lock is held so it's safe to access it
3623          * without res lock. See the description in ldlm_lock_decref_internal()
3624          * for the condition to free lvb_data of layout lock */
3625         if (lock->l_lvb_data != NULL) {
3626                 rc = obd_unpackmd(sbi->ll_dt_exp, &md.lsm,
3627                                   lock->l_lvb_data, lock->l_lvb_len);
3628                 if (rc >= 0) {
3629                         *gen = LL_LAYOUT_GEN_EMPTY;
3630                         if (md.lsm != NULL)
3631                                 *gen = md.lsm->lsm_layout_gen;
3632                         rc = 0;
3633                 } else {
3634                         CERROR("%s: file "DFID" unpackmd error: %d\n",
3635                                 ll_get_fsname(inode->i_sb, NULL, 0),
3636                                 PFID(&lli->lli_fid), rc);
3637                 }
3638         }
3639         if (rc < 0)
3640                 GOTO(out, rc);
3641
3642         /* set layout to file. Unlikely this will fail as old layout was
3643          * surely eliminated */
3644         memset(&conf, 0, sizeof conf);
3645         conf.coc_opc = OBJECT_CONF_SET;
3646         conf.coc_inode = inode;
3647         conf.coc_lock = lock;
3648         conf.u.coc_md = &md;
3649         rc = ll_layout_conf(inode, &conf);
3650
3651         if (md.lsm != NULL)
3652                 obd_free_memmd(sbi->ll_dt_exp, &md.lsm);
3653
3654         /* refresh layout failed, need to wait */
3655         wait_layout = rc == -EBUSY;
3656         EXIT;
3657
3658 out:
3659         LDLM_LOCK_PUT(lock);
3660         ldlm_lock_decref(lockh, mode);
3661
3662         /* wait for IO to complete if it's still being used. */
3663         if (wait_layout) {
3664                 CDEBUG(D_INODE, "%s: %p/"DFID" wait for layout reconf.\n",
3665                         ll_get_fsname(inode->i_sb, NULL, 0),
3666                         inode, PFID(&lli->lli_fid));
3667
3668                 memset(&conf, 0, sizeof conf);
3669                 conf.coc_opc = OBJECT_CONF_WAIT;
3670                 conf.coc_inode = inode;
3671                 rc = ll_layout_conf(inode, &conf);
3672                 if (rc == 0)
3673                         rc = -EAGAIN;
3674
3675                 CDEBUG(D_INODE, "file: "DFID" waiting layout return: %d.\n",
3676                         PFID(&lli->lli_fid), rc);
3677         }
3678         RETURN(rc);
3679 }
3680
3681 /**
3682  * This function checks if there exists a LAYOUT lock on the client side,
3683  * or enqueues it if it doesn't have one in cache.
3684  *
3685  * This function will not hold layout lock so it may be revoked any time after
3686  * this function returns. Any operations depend on layout should be redone
3687  * in that case.
3688  *
3689  * This function should be called before lov_io_init() to get an uptodate
3690  * layout version, the caller should save the version number and after IO
3691  * is finished, this function should be called again to verify that layout
3692  * is not changed during IO time.
3693  */
3694 int ll_layout_refresh(struct inode *inode, __u32 *gen)
3695 {
3696         struct ll_inode_info  *lli = ll_i2info(inode);
3697         struct ll_sb_info     *sbi = ll_i2sbi(inode);
3698         struct md_op_data     *op_data;
3699         struct lookup_intent   it;
3700         struct lustre_handle   lockh;
3701         ldlm_mode_t            mode;
3702         struct ldlm_enqueue_info einfo = {
3703                 .ei_type = LDLM_IBITS,
3704                 .ei_mode = LCK_CR,
3705                 .ei_cb_bl = ll_md_blocking_ast,
3706                 .ei_cb_cp = ldlm_completion_ast,
3707         };
3708         int rc;
3709         ENTRY;
3710
3711         *gen = lli->lli_layout_gen;
3712         if (!(sbi->ll_flags & LL_SBI_LAYOUT_LOCK))
3713                 RETURN(0);
3714
3715         /* sanity checks */
3716         LASSERT(fid_is_sane(ll_inode2fid(inode)));
3717         LASSERT(S_ISREG(inode->i_mode));
3718
3719         /* mostly layout lock is caching on the local side, so try to match
3720          * it before grabbing layout lock mutex. */
3721         mode = ll_take_md_lock(inode, MDS_INODELOCK_LAYOUT, &lockh, 0);
3722         if (mode != 0) { /* hit cached lock */
3723                 rc = ll_layout_lock_set(&lockh, mode, inode, gen, false);
3724                 if (rc == 0)
3725                         RETURN(0);
3726
3727                 /* better hold lli_layout_mutex to try again otherwise
3728                  * it will have starvation problem. */
3729         }
3730
3731         /* take layout lock mutex to enqueue layout lock exclusively. */
3732         mutex_lock(&lli->lli_layout_mutex);
3733
3734 again:
3735         /* try again. Maybe somebody else has done this. */
3736         mode = ll_take_md_lock(inode, MDS_INODELOCK_LAYOUT, &lockh, 0);
3737         if (mode != 0) { /* hit cached lock */
3738                 rc = ll_layout_lock_set(&lockh, mode, inode, gen, true);
3739                 if (rc == -EAGAIN)
3740                         goto again;
3741
3742                 mutex_unlock(&lli->lli_layout_mutex);
3743                 RETURN(rc);
3744         }
3745
3746         op_data = ll_prep_md_op_data(NULL, inode, inode, NULL,
3747                         0, 0, LUSTRE_OPC_ANY, NULL);
3748         if (IS_ERR(op_data)) {
3749                 mutex_unlock(&lli->lli_layout_mutex);
3750                 RETURN(PTR_ERR(op_data));
3751         }
3752
3753         /* have to enqueue one */
3754         memset(&it, 0, sizeof(it));
3755         it.it_op = IT_LAYOUT;
3756         lockh.cookie = 0ULL;
3757
3758         LDLM_DEBUG_NOLOCK("%s: requeue layout lock for file %p/"DFID".\n",
3759                         ll_get_fsname(inode->i_sb, NULL, 0), inode,
3760                         PFID(&lli->lli_fid));
3761
3762         rc = md_enqueue(sbi->ll_md_exp, &einfo, &it, op_data, &lockh,
3763                         NULL, 0, NULL, 0);
3764         if (it.d.lustre.it_data != NULL)
3765                 ptlrpc_req_finished(it.d.lustre.it_data);
3766         it.d.lustre.it_data = NULL;
3767
3768         ll_finish_md_op_data(op_data);
3769
3770         mode = it.d.lustre.it_lock_mode;
3771         it.d.lustre.it_lock_mode = 0;
3772         ll_intent_drop_lock(&it);
3773
3774         if (rc == 0) {
3775                 /* set lock data in case this is a new lock */
3776                 ll_set_lock_data(sbi->ll_md_exp, inode, &it, NULL);
3777                 rc = ll_layout_lock_set(&lockh, mode, inode, gen, true);
3778                 if (rc == -EAGAIN)
3779                         goto again;
3780         }
3781         mutex_unlock(&lli->lli_layout_mutex);
3782
3783         RETURN(rc);
3784 }
3785
3786 /**
3787  *  This function send a restore request to the MDT
3788  */
3789 int ll_layout_restore(struct inode *inode)
3790 {
3791         struct hsm_user_request *hur;
3792         int                      len, rc;
3793         ENTRY;
3794
3795         len = sizeof(struct hsm_user_request) +
3796               sizeof(struct hsm_user_item);
3797         OBD_ALLOC(hur, len);
3798         if (hur == NULL)
3799                 RETURN(-ENOMEM);
3800
3801         hur->hur_request.hr_action = HUA_RESTORE;
3802         hur->hur_request.hr_archive_id = 0;
3803         hur->hur_request.hr_flags = 0;
3804         memcpy(&hur->hur_user_item[0].hui_fid, &ll_i2info(inode)->lli_fid,
3805                sizeof(hur->hur_user_item[0].hui_fid));
3806         hur->hur_user_item[0].hui_extent.length = -1;
3807         hur->hur_request.hr_itemcount = 1;
3808         rc = obd_iocontrol(LL_IOC_HSM_REQUEST, cl_i2sbi(inode)->ll_md_exp,
3809                            len, hur, NULL);
3810         OBD_FREE(hur, len);
3811         RETURN(rc);
3812 }
3813