Merge remote-tracking branch 'staging/staging-next'

[deliverable/linux.git] / drivers / staging / lustre / lustre / llite / file.c

/*
 * GPL HEADER START
 *
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 2 only,
 * as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License version 2 for more details (a copy is included
 * in the LICENSE file that accompanied this code).
 *
 * You should have received a copy of the GNU General Public License
 * version 2 along with this program; If not, see
 * http://www.gnu.org/licenses/gpl-2.0.html
 *
 * GPL HEADER END
 */
/*
 * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
 * Use is subject to license terms.
 *
 * Copyright (c) 2011, 2015, Intel Corporation.
 */
/*
 * This file is part of Lustre, http://www.lustre.org/
 * Lustre is a trademark of Sun Microsystems, Inc.
 *
 * lustre/llite/file.c
 *
 * Author: Peter Braam <braam@clusterfs.com>
 * Author: Phil Schwan <phil@clusterfs.com>
 * Author: Andreas Dilger <adilger@clusterfs.com>
 */

#define DEBUG_SUBSYSTEM S_LLITE
#include "../include/lustre_dlm.h"
#include "../include/lustre_lite.h"
#include <linux/pagemap.h>
#include <linux/file.h>
#include <linux/sched.h>
#include <linux/mount.h>
#include "llite_internal.h"
#include "../include/lustre/ll_fiemap.h"
#include "../include/lustre/lustre_ioctl.h"

#include "../include/cl_object.h"

static int
ll_put_grouplock(struct inode *inode, struct file *file, unsigned long arg);

static int ll_lease_close(struct obd_client_handle *och, struct inode *inode,
                          bool *lease_broken);

static enum llioc_iter
ll_iocontrol_call(struct inode *inode, struct file *file,
                  unsigned int cmd, unsigned long arg, int *rcp);

static struct ll_file_data *ll_file_data_get(void)
{
        struct ll_file_data *fd;

        fd = kmem_cache_zalloc(ll_file_data_slab, GFP_NOFS);
        if (!fd)
                return NULL;
        fd->fd_write_failed = false;
        return fd;
}

static void ll_file_data_put(struct ll_file_data *fd)
{
        if (fd)
                kmem_cache_free(ll_file_data_slab, fd);
}

void ll_pack_inode2opdata(struct inode *inode, struct md_op_data *op_data,
                          struct lustre_handle *fh)
{
        op_data->op_fid1 = ll_i2info(inode)->lli_fid;
        op_data->op_attr.ia_mode = inode->i_mode;
        op_data->op_attr.ia_atime = inode->i_atime;
        op_data->op_attr.ia_mtime = inode->i_mtime;
        op_data->op_attr.ia_ctime = inode->i_ctime;
        op_data->op_attr.ia_size = i_size_read(inode);
        op_data->op_attr_blocks = inode->i_blocks;
        op_data->op_attr_flags = ll_inode_to_ext_flags(inode->i_flags);
        op_data->op_ioepoch = ll_i2info(inode)->lli_ioepoch;
        if (fh)
                op_data->op_handle = *fh;

        if (ll_i2info(inode)->lli_flags & LLIF_DATA_MODIFIED)
                op_data->op_bias |= MDS_DATA_MODIFIED;
}

/**
 * Closes the IO epoch and packs all the attributes into @op_data for
 * the CLOSE rpc.
 */
static void ll_prepare_close(struct inode *inode, struct md_op_data *op_data,
                             struct obd_client_handle *och)
{
        op_data->op_attr.ia_valid = ATTR_MODE | ATTR_ATIME | ATTR_ATIME_SET |
                                        ATTR_MTIME | ATTR_MTIME_SET |
                                        ATTR_CTIME | ATTR_CTIME_SET;

        if (!(och->och_flags & FMODE_WRITE))
                goto out;

        if (!exp_connect_som(ll_i2mdexp(inode)) || !S_ISREG(inode->i_mode))
                op_data->op_attr.ia_valid |= ATTR_SIZE | ATTR_BLOCKS;
        else
                ll_ioepoch_close(inode, op_data, &och, 0);

out:
        ll_pack_inode2opdata(inode, op_data, &och->och_fh);
        ll_prep_md_op_data(op_data, inode, NULL, NULL,
                           0, 0, LUSTRE_OPC_ANY, NULL);
}

static int ll_close_inode_openhandle(struct obd_export *md_exp,
                                     struct inode *inode,
                                     struct obd_client_handle *och,
                                     const __u64 *data_version)
{
        struct obd_export *exp = ll_i2mdexp(inode);
        struct md_op_data *op_data;
        struct ptlrpc_request *req = NULL;
        struct obd_device *obd = class_exp2obd(exp);
        int epoch_close = 1;
        int rc;

        if (!obd) {
                /*
                 * XXX: in case of LMV, is this correct to access
                 * ->exp_handle?
                 */
                CERROR("Invalid MDC connection handle %#llx\n",
                       ll_i2mdexp(inode)->exp_handle.h_cookie);
                rc = 0;
                goto out;
        }

        op_data = kzalloc(sizeof(*op_data), GFP_NOFS);
        if (!op_data) {
                /* XXX We leak openhandle and request here. */
                rc = -ENOMEM;
                goto out;
        }

        ll_prepare_close(inode, op_data, och);
        if (data_version) {
                /* Pass in data_version implies release. */
                op_data->op_bias |= MDS_HSM_RELEASE;
                op_data->op_data_version = *data_version;
                op_data->op_lease_handle = och->och_lease_handle;
                op_data->op_attr.ia_valid |= ATTR_SIZE | ATTR_BLOCKS;
        }
        epoch_close = op_data->op_flags & MF_EPOCH_CLOSE;
        rc = md_close(md_exp, op_data, och->och_mod, &req);
        if (rc == -EAGAIN) {
                /* This close must have the epoch closed. */
                LASSERT(epoch_close);
                /* MDS has instructed us to obtain Size-on-MDS attribute from
                 * OSTs and send setattr to back to MDS.
                 */
                rc = ll_som_update(inode, op_data);
                if (rc) {
                        CERROR("%s: inode "DFID" mdc Size-on-MDS update failed: rc = %d\n",
                               ll_i2mdexp(inode)->exp_obd->obd_name,
                               PFID(ll_inode2fid(inode)), rc);
                        rc = 0;
                }
        } else if (rc) {
                CERROR("%s: inode "DFID" mdc close failed: rc = %d\n",
                       ll_i2mdexp(inode)->exp_obd->obd_name,
                       PFID(ll_inode2fid(inode)), rc);
        }

        /* DATA_MODIFIED flag was successfully sent on close, cancel data
         * modification flag.
         */
        if (rc == 0 && (op_data->op_bias & MDS_DATA_MODIFIED)) {
                struct ll_inode_info *lli = ll_i2info(inode);

                spin_lock(&lli->lli_lock);
                lli->lli_flags &= ~LLIF_DATA_MODIFIED;
                spin_unlock(&lli->lli_lock);
        }

        if (rc == 0) {
                rc = ll_objects_destroy(req, inode);
                if (rc)
                        CERROR("inode %lu ll_objects destroy: rc = %d\n",
                               inode->i_ino, rc);
        }
        if (rc == 0 && op_data->op_bias & MDS_HSM_RELEASE) {
                struct mdt_body *body;

                body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
                if (!(body->mbo_valid & OBD_MD_FLRELEASED))
                        rc = -EBUSY;
        }

        ll_finish_md_op_data(op_data);

out:
        if (exp_connect_som(exp) && !epoch_close &&
            S_ISREG(inode->i_mode) && (och->och_flags & FMODE_WRITE)) {
                ll_queue_done_writing(inode, LLIF_DONE_WRITING);
        } else {
                md_clear_open_replay_data(md_exp, och);
                /* Free @och if it is not waiting for DONE_WRITING. */
                och->och_fh.cookie = DEAD_HANDLE_MAGIC;
                kfree(och);
        }
        if (req) /* This is close request */
                ptlrpc_req_finished(req);
        return rc;
}

int ll_md_real_close(struct inode *inode, fmode_t fmode)
{
        struct ll_inode_info *lli = ll_i2info(inode);
        struct obd_client_handle **och_p;
        struct obd_client_handle *och;
        __u64 *och_usecount;
        int rc = 0;

        if (fmode & FMODE_WRITE) {
                och_p = &lli->lli_mds_write_och;
                och_usecount = &lli->lli_open_fd_write_count;
        } else if (fmode & FMODE_EXEC) {
                och_p = &lli->lli_mds_exec_och;
                och_usecount = &lli->lli_open_fd_exec_count;
        } else {
                LASSERT(fmode & FMODE_READ);
                och_p = &lli->lli_mds_read_och;
                och_usecount = &lli->lli_open_fd_read_count;
        }

        mutex_lock(&lli->lli_och_mutex);
        if (*och_usecount > 0) {
                /* There are still users of this handle, so skip
                 * freeing it.
                 */
                mutex_unlock(&lli->lli_och_mutex);
                return 0;
        }

        och = *och_p;
        *och_p = NULL;
        mutex_unlock(&lli->lli_och_mutex);

        if (och) {
                /* There might be a race and this handle may already
                 * be closed.
                 */
                rc = ll_close_inode_openhandle(ll_i2sbi(inode)->ll_md_exp,
                                               inode, och, NULL);
        }

        return rc;
}

static int ll_md_close(struct obd_export *md_exp, struct inode *inode,
                       struct file *file)
{
        struct ll_file_data *fd = LUSTRE_FPRIVATE(file);
        struct ll_inode_info *lli = ll_i2info(inode);
        int lockmode;
        __u64 flags = LDLM_FL_BLOCK_GRANTED | LDLM_FL_TEST_LOCK;
        struct lustre_handle lockh;
        ldlm_policy_data_t policy = {.l_inodebits = {MDS_INODELOCK_OPEN} };
        int rc = 0;

        /* clear group lock, if present */
        if (unlikely(fd->fd_flags & LL_FILE_GROUP_LOCKED))
                ll_put_grouplock(inode, file, fd->fd_grouplock.lg_gid);

        if (fd->fd_lease_och) {
                bool lease_broken;

                /* Usually the lease is not released when the
                 * application crashed, we need to release here.
                 */
                rc = ll_lease_close(fd->fd_lease_och, inode, &lease_broken);
                CDEBUG(rc ? D_ERROR : D_INODE,
                       "Clean up lease " DFID " %d/%d\n",
                       PFID(&lli->lli_fid), rc, lease_broken);

                fd->fd_lease_och = NULL;
        }

        if (fd->fd_och) {
                rc = ll_close_inode_openhandle(md_exp, inode, fd->fd_och, NULL);
                fd->fd_och = NULL;
                goto out;
        }

        /* Let's see if we have good enough OPEN lock on the file and if
         * we can skip talking to MDS
         */

        mutex_lock(&lli->lli_och_mutex);
        if (fd->fd_omode & FMODE_WRITE) {
                lockmode = LCK_CW;
                LASSERT(lli->lli_open_fd_write_count);
                lli->lli_open_fd_write_count--;
        } else if (fd->fd_omode & FMODE_EXEC) {
                lockmode = LCK_PR;
                LASSERT(lli->lli_open_fd_exec_count);
                lli->lli_open_fd_exec_count--;
        } else {
                lockmode = LCK_CR;
                LASSERT(lli->lli_open_fd_read_count);
                lli->lli_open_fd_read_count--;
        }
        mutex_unlock(&lli->lli_och_mutex);

        if (!md_lock_match(md_exp, flags, ll_inode2fid(inode),
                           LDLM_IBITS, &policy, lockmode, &lockh))
                rc = ll_md_real_close(inode, fd->fd_omode);

out:
        LUSTRE_FPRIVATE(file) = NULL;
        ll_file_data_put(fd);

        return rc;
}

/* While this returns an error code, fput() the caller does not, so we need
 * to make every effort to clean up all of our state here.  Also, applications
 * rarely check close errors and even if an error is returned they will not
 * re-try the close call.
 */
int ll_file_release(struct inode *inode, struct file *file)
{
        struct ll_file_data *fd;
        struct ll_sb_info *sbi = ll_i2sbi(inode);
        struct ll_inode_info *lli = ll_i2info(inode);
        int rc;

        CDEBUG(D_VFSTRACE, "VFS Op:inode="DFID"(%p)\n",
               PFID(ll_inode2fid(inode)), inode);

        if (!is_root_inode(inode))
                ll_stats_ops_tally(sbi, LPROC_LL_RELEASE, 1);
        fd = LUSTRE_FPRIVATE(file);
        LASSERT(fd);

        /* The last ref on @file, maybe not be the owner pid of statahead.
         * Different processes can open the same dir, "ll_opendir_key" means:
         * it is me that should stop the statahead thread.
         */
        if (S_ISDIR(inode->i_mode) && lli->lli_opendir_key == fd &&
            lli->lli_opendir_pid != 0)
                ll_stop_statahead(inode, lli->lli_opendir_key);

        if (is_root_inode(inode)) {
                LUSTRE_FPRIVATE(file) = NULL;
                ll_file_data_put(fd);
                return 0;
        }

        if (!S_ISDIR(inode->i_mode)) {
                if (lli->lli_clob)
                        lov_read_and_clear_async_rc(lli->lli_clob);
                lli->lli_async_rc = 0;
        }

        rc = ll_md_close(sbi->ll_md_exp, inode, file);

        if (CFS_FAIL_TIMEOUT_MS(OBD_FAIL_PTLRPC_DUMP_LOG, cfs_fail_val))
                libcfs_debug_dumplog();

        return rc;
}

static int ll_intent_file_open(struct dentry *de, void *lmm, int lmmsize,
                               struct lookup_intent *itp)
{
        struct inode *inode = d_inode(de);
        struct ll_sb_info *sbi = ll_i2sbi(inode);
        struct dentry *parent = de->d_parent;
        const char *name = NULL;
        struct md_op_data *op_data;
        struct ptlrpc_request *req = NULL;
        int len = 0, rc;

        LASSERT(parent);
        LASSERT(itp->it_flags & MDS_OPEN_BY_FID);

        /*
         * if server supports open-by-fid, or file name is invalid, don't pack
         * name in open request
         */
        if (!(exp_connect_flags(sbi->ll_md_exp) & OBD_CONNECT_OPEN_BY_FID) &&
            lu_name_is_valid_2(de->d_name.name, de->d_name.len)) {
                name = de->d_name.name;
                len = de->d_name.len;
        }

        op_data  = ll_prep_md_op_data(NULL, d_inode(parent), inode, name, len,
                                      O_RDWR, LUSTRE_OPC_ANY, NULL);
        if (IS_ERR(op_data))
                return PTR_ERR(op_data);
        op_data->op_data = lmm;
        op_data->op_data_size = lmmsize;

        rc = md_intent_lock(sbi->ll_md_exp, op_data, itp, &req,
                            &ll_md_blocking_ast, 0);
        ll_finish_md_op_data(op_data);
        if (rc == -ESTALE) {
                /* reason for keep own exit path - don`t flood log
                * with messages with -ESTALE errors.
                */
                if (!it_disposition(itp, DISP_OPEN_OPEN) ||
                    it_open_error(DISP_OPEN_OPEN, itp))
                        goto out;
                ll_release_openhandle(inode, itp);
                goto out;
        }

        if (it_disposition(itp, DISP_LOOKUP_NEG)) {
                rc = -ENOENT;
                goto out;
        }

        if (rc != 0 || it_open_error(DISP_OPEN_OPEN, itp)) {
                rc = rc ? rc : it_open_error(DISP_OPEN_OPEN, itp);
                CDEBUG(D_VFSTRACE, "lock enqueue: err: %d\n", rc);
                goto out;
        }

        rc = ll_prep_inode(&inode, req, NULL, itp);
        if (!rc && itp->it_lock_mode)
                ll_set_lock_data(sbi->ll_md_exp, inode, itp, NULL);

out:
        ptlrpc_req_finished(req);
        ll_intent_drop_lock(itp);

        return rc;
}

/**
 * Assign an obtained @ioepoch to client's inode. No lock is needed, MDS does
 * not believe attributes if a few ioepoch holders exist. Attributes for
 * previous ioepoch if new one is opened are also skipped by MDS.
 */
void ll_ioepoch_open(struct ll_inode_info *lli, __u64 ioepoch)
{
        if (ioepoch && lli->lli_ioepoch != ioepoch) {
                lli->lli_ioepoch = ioepoch;
                CDEBUG(D_INODE, "Epoch %llu opened on "DFID"\n",
                       ioepoch, PFID(&lli->lli_fid));
        }
}

static int ll_och_fill(struct obd_export *md_exp, struct lookup_intent *it,
                       struct obd_client_handle *och)
{
        struct mdt_body *body;

        body = req_capsule_server_get(&it->it_request->rq_pill, &RMF_MDT_BODY);
        och->och_fh = body->mbo_handle;
        och->och_fid = body->mbo_fid1;
        och->och_lease_handle.cookie = it->it_lock_handle;
        och->och_magic = OBD_CLIENT_HANDLE_MAGIC;
        och->och_flags = it->it_flags;

        return md_set_open_replay_data(md_exp, och, it);
}

static int ll_local_open(struct file *file, struct lookup_intent *it,
                         struct ll_file_data *fd, struct obd_client_handle *och)
{
        struct inode *inode = file_inode(file);
        struct ll_inode_info *lli = ll_i2info(inode);

        LASSERT(!LUSTRE_FPRIVATE(file));

        LASSERT(fd);

        if (och) {
                struct mdt_body *body;
                int rc;

                rc = ll_och_fill(ll_i2sbi(inode)->ll_md_exp, it, och);
                if (rc != 0)
                        return rc;

                body = req_capsule_server_get(&it->it_request->rq_pill,
                                              &RMF_MDT_BODY);
                ll_ioepoch_open(lli, body->mbo_ioepoch);
        }

        LUSTRE_FPRIVATE(file) = fd;
        ll_readahead_init(inode, &fd->fd_ras);
        fd->fd_omode = it->it_flags & (FMODE_READ | FMODE_WRITE | FMODE_EXEC);

        /* ll_cl_context initialize */
        rwlock_init(&fd->fd_lock);
        INIT_LIST_HEAD(&fd->fd_lccs);

        return 0;
}

/* Open a file, and (for the very first open) create objects on the OSTs at
 * this time.  If opened with O_LOV_DELAY_CREATE, then we don't do the object
 * creation or open until ll_lov_setstripe() ioctl is called.
 *
 * If we already have the stripe MD locally then we don't request it in
 * md_open(), by passing a lmm_size = 0.
 *
 * It is up to the application to ensure no other processes open this file
 * in the O_LOV_DELAY_CREATE case, or the default striping pattern will be
 * used.  We might be able to avoid races of that sort by getting lli_open_sem
 * before returning in the O_LOV_DELAY_CREATE case and dropping it here
 * or in ll_file_release(), but I'm not sure that is desirable/necessary.
 */
int ll_file_open(struct inode *inode, struct file *file)
{
        struct ll_inode_info *lli = ll_i2info(inode);
        struct lookup_intent *it, oit = { .it_op = IT_OPEN,
                                          .it_flags = file->f_flags };
        struct obd_client_handle **och_p = NULL;
        __u64 *och_usecount = NULL;
        struct ll_file_data *fd;
        int rc = 0, opendir_set = 0;

        CDEBUG(D_VFSTRACE, "VFS Op:inode="DFID"(%p), flags %o\n",
               PFID(ll_inode2fid(inode)), inode, file->f_flags);

        it = file->private_data; /* XXX: compat macro */
        file->private_data = NULL; /* prevent ll_local_open assertion */

        fd = ll_file_data_get();
        if (!fd) {
                rc = -ENOMEM;
                goto out_openerr;
        }

        fd->fd_file = file;
        if (S_ISDIR(inode->i_mode)) {
                spin_lock(&lli->lli_sa_lock);
                if (!lli->lli_opendir_key && !lli->lli_sai &&
                    lli->lli_opendir_pid == 0) {
                        lli->lli_opendir_key = fd;
                        lli->lli_opendir_pid = current_pid();
                        opendir_set = 1;
                }
                spin_unlock(&lli->lli_sa_lock);
        }

        if (is_root_inode(inode)) {
                LUSTRE_FPRIVATE(file) = fd;
                return 0;
        }

        if (!it || !it->it_disposition) {
                /* Convert f_flags into access mode. We cannot use file->f_mode,
                 * because everything but O_ACCMODE mask was stripped from
                 * there
                 */
                if ((oit.it_flags + 1) & O_ACCMODE)
                        oit.it_flags++;
                if (file->f_flags & O_TRUNC)
                        oit.it_flags |= FMODE_WRITE;

                /* kernel only call f_op->open in dentry_open.  filp_open calls
                 * dentry_open after call to open_namei that checks permissions.
                 * Only nfsd_open call dentry_open directly without checking
                 * permissions and because of that this code below is safe.
                 */
                if (oit.it_flags & (FMODE_WRITE | FMODE_READ))
                        oit.it_flags |= MDS_OPEN_OWNEROVERRIDE;

                /* We do not want O_EXCL here, presumably we opened the file
                 * already? XXX - NFS implications?
                 */
                oit.it_flags &= ~O_EXCL;

                /* bug20584, if "it_flags" contains O_CREAT, the file will be
                 * created if necessary, then "IT_CREAT" should be set to keep
                 * consistent with it
                 */
                if (oit.it_flags & O_CREAT)
                        oit.it_op |= IT_CREAT;

                it = &oit;
        }

restart:
        /* Let's see if we have file open on MDS already. */
        if (it->it_flags & FMODE_WRITE) {
                och_p = &lli->lli_mds_write_och;
                och_usecount = &lli->lli_open_fd_write_count;
        } else if (it->it_flags & FMODE_EXEC) {
                och_p = &lli->lli_mds_exec_och;
                och_usecount = &lli->lli_open_fd_exec_count;
         } else {
                och_p = &lli->lli_mds_read_och;
                och_usecount = &lli->lli_open_fd_read_count;
        }

        mutex_lock(&lli->lli_och_mutex);
        if (*och_p) { /* Open handle is present */
                if (it_disposition(it, DISP_OPEN_OPEN)) {
                        /* Well, there's extra open request that we do not need,
                         * let's close it somehow. This will decref request.
                         */
                        rc = it_open_error(DISP_OPEN_OPEN, it);
                        if (rc) {
                                mutex_unlock(&lli->lli_och_mutex);
                                goto out_openerr;
                        }

                        ll_release_openhandle(inode, it);
                }
                (*och_usecount)++;

                rc = ll_local_open(file, it, fd, NULL);
                if (rc) {
                        (*och_usecount)--;
                        mutex_unlock(&lli->lli_och_mutex);
                        goto out_openerr;
                }
        } else {
                LASSERT(*och_usecount == 0);
                if (!it->it_disposition) {
                        /* We cannot just request lock handle now, new ELC code
                         * means that one of other OPEN locks for this file
                         * could be cancelled, and since blocking ast handler
                         * would attempt to grab och_mutex as well, that would
                         * result in a deadlock
                         */
                        mutex_unlock(&lli->lli_och_mutex);
                        /*
                         * Normally called under two situations:
                         * 1. NFS export.
                         * 2. revalidate with IT_OPEN (revalidate doesn't
                         *    execute this intent any more).
                         *
                         * Always fetch MDS_OPEN_LOCK if this is not setstripe.
                         *
                         * Always specify MDS_OPEN_BY_FID because we don't want
                         * to get file with different fid.
                         */
                        it->it_flags |= MDS_OPEN_LOCK | MDS_OPEN_BY_FID;
                        rc = ll_intent_file_open(file->f_path.dentry, NULL, 0, it);
                        if (rc)
                                goto out_openerr;

                        goto restart;
                }
                *och_p = kzalloc(sizeof(struct obd_client_handle), GFP_NOFS);
                if (!*och_p) {
                        rc = -ENOMEM;
                        goto out_och_free;
                }

                (*och_usecount)++;

                /* md_intent_lock() didn't get a request ref if there was an
                 * open error, so don't do cleanup on the request here
                 * (bug 3430)
                 */
                /* XXX (green): Should not we bail out on any error here, not
                 * just open error?
                 */
                rc = it_open_error(DISP_OPEN_OPEN, it);
                if (rc)
                        goto out_och_free;

                LASSERTF(it_disposition(it, DISP_ENQ_OPEN_REF),
                         "inode %p: disposition %x, status %d\n", inode,
                         it_disposition(it, ~0), it->it_status);

                rc = ll_local_open(file, it, fd, *och_p);
                if (rc)
                        goto out_och_free;
        }
        mutex_unlock(&lli->lli_och_mutex);
        fd = NULL;

        /* Must do this outside lli_och_mutex lock to prevent deadlock where
         * different kind of OPEN lock for this same inode gets cancelled
         * by ldlm_cancel_lru
         */
        if (!S_ISREG(inode->i_mode))
                goto out_och_free;

        if (!lli->lli_has_smd &&
            (cl_is_lov_delay_create(file->f_flags) ||
             (file->f_mode & FMODE_WRITE) == 0)) {
                CDEBUG(D_INODE, "object creation was delayed\n");
                goto out_och_free;
        }
        cl_lov_delay_create_clear(&file->f_flags);
        goto out_och_free;

out_och_free:
        if (rc) {
                if (och_p && *och_p) {
                        kfree(*och_p);
                        *och_p = NULL;
                        (*och_usecount)--;
                }
                mutex_unlock(&lli->lli_och_mutex);

out_openerr:
                if (opendir_set != 0)
                        ll_stop_statahead(inode, lli->lli_opendir_key);
                ll_file_data_put(fd);
        } else {
                ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_OPEN, 1);
        }

        if (it && it_disposition(it, DISP_ENQ_OPEN_REF)) {
                ptlrpc_req_finished(it->it_request);
                it_clear_disposition(it, DISP_ENQ_OPEN_REF);
        }

        return rc;
}

static int ll_md_blocking_lease_ast(struct ldlm_lock *lock,
                                    struct ldlm_lock_desc *desc,
                                    void *data, int flag)
{
        int rc;
        struct lustre_handle lockh;

        switch (flag) {
        case LDLM_CB_BLOCKING:
                ldlm_lock2handle(lock, &lockh);
                rc = ldlm_cli_cancel(&lockh, LCF_ASYNC);
                if (rc < 0) {
                        CDEBUG(D_INODE, "ldlm_cli_cancel: %d\n", rc);
                        return rc;
                }
                break;
        case LDLM_CB_CANCELING:
                /* do nothing */
                break;
        }
        return 0;
}

/**
 * Acquire a lease and open the file.
 */
static struct obd_client_handle *
ll_lease_open(struct inode *inode, struct file *file, fmode_t fmode,
              __u64 open_flags)
{
        struct lookup_intent it = { .it_op = IT_OPEN };
        struct ll_sb_info *sbi = ll_i2sbi(inode);
        struct md_op_data *op_data;
        struct ptlrpc_request *req = NULL;
        struct lustre_handle old_handle = { 0 };
        struct obd_client_handle *och = NULL;
        int rc;
        int rc2;

        if (fmode != FMODE_WRITE && fmode != FMODE_READ)
                return ERR_PTR(-EINVAL);

        if (file) {
                struct ll_inode_info *lli = ll_i2info(inode);
                struct ll_file_data *fd = LUSTRE_FPRIVATE(file);
                struct obd_client_handle **och_p;
                __u64 *och_usecount;

                if (!(fmode & file->f_mode) || (file->f_mode & FMODE_EXEC))
                        return ERR_PTR(-EPERM);

                /* Get the openhandle of the file */
                rc = -EBUSY;
                mutex_lock(&lli->lli_och_mutex);
                if (fd->fd_lease_och) {
                        mutex_unlock(&lli->lli_och_mutex);
                        return ERR_PTR(rc);
                }

                if (!fd->fd_och) {
                        if (file->f_mode & FMODE_WRITE) {
                                LASSERT(lli->lli_mds_write_och);
                                och_p = &lli->lli_mds_write_och;
                                och_usecount = &lli->lli_open_fd_write_count;
                        } else {
                                LASSERT(lli->lli_mds_read_och);
                                och_p = &lli->lli_mds_read_och;
                                och_usecount = &lli->lli_open_fd_read_count;
                        }
                        if (*och_usecount == 1) {
                                fd->fd_och = *och_p;
                                *och_p = NULL;
                                *och_usecount = 0;
                                rc = 0;
                        }
                }
                mutex_unlock(&lli->lli_och_mutex);
                if (rc < 0) /* more than 1 opener */
                        return ERR_PTR(rc);

                LASSERT(fd->fd_och);
                old_handle = fd->fd_och->och_fh;
        }

        och = kzalloc(sizeof(*och), GFP_NOFS);
        if (!och)
                return ERR_PTR(-ENOMEM);

        op_data = ll_prep_md_op_data(NULL, inode, inode, NULL, 0, 0,
                                     LUSTRE_OPC_ANY, NULL);
        if (IS_ERR(op_data)) {
                rc = PTR_ERR(op_data);
                goto out;
        }

        /* To tell the MDT this openhandle is from the same owner */
        op_data->op_handle = old_handle;

        it.it_flags = fmode | open_flags;
        it.it_flags |= MDS_OPEN_LOCK | MDS_OPEN_BY_FID | MDS_OPEN_LEASE;
        rc = md_intent_lock(sbi->ll_md_exp, op_data, &it, &req,
                            &ll_md_blocking_lease_ast,
        /* LDLM_FL_NO_LRU: To not put the lease lock into LRU list, otherwise
         * it can be cancelled which may mislead applications that the lease is
         * broken;
         * LDLM_FL_EXCL: Set this flag so that it won't be matched by normal
         * open in ll_md_blocking_ast(). Otherwise as ll_md_blocking_lease_ast
         * doesn't deal with openhandle, so normal openhandle will be leaked.
         */
                            LDLM_FL_NO_LRU | LDLM_FL_EXCL);
        ll_finish_md_op_data(op_data);
        ptlrpc_req_finished(req);
        if (rc < 0)
                goto out_release_it;

        if (it_disposition(&it, DISP_LOOKUP_NEG)) {
                rc = -ENOENT;
                goto out_release_it;
        }

        rc = it_open_error(DISP_OPEN_OPEN, &it);
        if (rc)
                goto out_release_it;

        LASSERT(it_disposition(&it, DISP_ENQ_OPEN_REF));
        ll_och_fill(sbi->ll_md_exp, &it, och);

        if (!it_disposition(&it, DISP_OPEN_LEASE)) /* old server? */ {
                rc = -EOPNOTSUPP;
                goto out_close;
        }

        /* already get lease, handle lease lock */
        ll_set_lock_data(sbi->ll_md_exp, inode, &it, NULL);
        if (it.it_lock_mode == 0 ||
            it.it_lock_bits != MDS_INODELOCK_OPEN) {
                /* open lock must return for lease */
                CERROR(DFID "lease granted but no open lock, %d/%llu.\n",
                       PFID(ll_inode2fid(inode)), it.it_lock_mode,
                       it.it_lock_bits);
                rc = -EPROTO;
                goto out_close;
        }

        ll_intent_release(&it);
        return och;

out_close:
        /* Cancel open lock */
        if (it.it_lock_mode != 0) {
                ldlm_lock_decref_and_cancel(&och->och_lease_handle,
                                            it.it_lock_mode);
                it.it_lock_mode = 0;
                och->och_lease_handle.cookie = 0ULL;
        }
        rc2 = ll_close_inode_openhandle(sbi->ll_md_exp, inode, och, NULL);
        if (rc2 < 0)
                CERROR("%s: error closing file "DFID": %d\n",
                       ll_get_fsname(inode->i_sb, NULL, 0),
                       PFID(&ll_i2info(inode)->lli_fid), rc2);
        och = NULL; /* och has been freed in ll_close_inode_openhandle() */
out_release_it:
        ll_intent_release(&it);
out:
        kfree(och);
        return ERR_PTR(rc);
}

/**
 * Release lease and close the file.
 * It will check if the lease has ever broken.
 */
static int ll_lease_close(struct obd_client_handle *och, struct inode *inode,
                          bool *lease_broken)
{
        struct ldlm_lock *lock;
        bool cancelled = true;
        int rc;

        lock = ldlm_handle2lock(&och->och_lease_handle);
        if (lock) {
                lock_res_and_lock(lock);
                cancelled = ldlm_is_cancel(lock);
                unlock_res_and_lock(lock);
                LDLM_LOCK_PUT(lock);
        }

        CDEBUG(D_INODE, "lease for " DFID " broken? %d\n",
               PFID(&ll_i2info(inode)->lli_fid), cancelled);

        if (!cancelled)
                ldlm_cli_cancel(&och->och_lease_handle, 0);
        if (lease_broken)
                *lease_broken = cancelled;

        rc = ll_close_inode_openhandle(ll_i2sbi(inode)->ll_md_exp, inode, och,
                                       NULL);
        return rc;
}

/* Fills the obdo with the attributes for the lsm */
static int ll_lsm_getattr(struct lov_stripe_md *lsm, struct obd_export *exp,
                          struct obdo *obdo, __u64 ioepoch, int dv_flags)
{
        struct ptlrpc_request_set *set;
        struct obd_info     oinfo = { };
        int                     rc;

        LASSERT(lsm);

        oinfo.oi_md = lsm;
        oinfo.oi_oa = obdo;
        oinfo.oi_oa->o_oi = lsm->lsm_oi;
        oinfo.oi_oa->o_mode = S_IFREG;
        oinfo.oi_oa->o_ioepoch = ioepoch;
        oinfo.oi_oa->o_valid = OBD_MD_FLID | OBD_MD_FLTYPE |
                               OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
                               OBD_MD_FLBLKSZ | OBD_MD_FLATIME |
                               OBD_MD_FLMTIME | OBD_MD_FLCTIME |
                               OBD_MD_FLGROUP | OBD_MD_FLEPOCH |
                               OBD_MD_FLDATAVERSION;
        if (dv_flags & (LL_DV_WR_FLUSH | LL_DV_RD_FLUSH)) {
                oinfo.oi_oa->o_valid |= OBD_MD_FLFLAGS;
                oinfo.oi_oa->o_flags |= OBD_FL_SRVLOCK;
                if (dv_flags & LL_DV_WR_FLUSH)
                        oinfo.oi_oa->o_flags |= OBD_FL_FLUSH;
        }

        set = ptlrpc_prep_set();
        if (!set) {
                CERROR("cannot allocate ptlrpc set: rc = %d\n", -ENOMEM);
                rc = -ENOMEM;
        } else {
                rc = obd_getattr_async(exp, &oinfo, set);
                if (rc == 0)
                        rc = ptlrpc_set_wait(set);
                ptlrpc_set_destroy(set);
        }
        if (rc == 0) {
                oinfo.oi_oa->o_valid &= (OBD_MD_FLBLOCKS | OBD_MD_FLBLKSZ |
                                         OBD_MD_FLATIME | OBD_MD_FLMTIME |
                                         OBD_MD_FLCTIME | OBD_MD_FLSIZE |
                                         OBD_MD_FLDATAVERSION | OBD_MD_FLFLAGS);
                if (dv_flags & LL_DV_WR_FLUSH &&
                    !(oinfo.oi_oa->o_valid & OBD_MD_FLFLAGS &&
                      oinfo.oi_oa->o_flags & OBD_FL_FLUSH))
                        return -ENOTSUPP;
        }
        return rc;
}

/**
  * Performs the getattr on the inode and updates its fields.
  * If @sync != 0, perform the getattr under the server-side lock.
  */
int ll_inode_getattr(struct inode *inode, struct obdo *obdo,
                     __u64 ioepoch, int sync)
{
        struct lov_stripe_md *lsm;
        int rc;

        lsm = ccc_inode_lsm_get(inode);
        rc = ll_lsm_getattr(lsm, ll_i2dtexp(inode),
                            obdo, ioepoch, sync ? LL_DV_RD_FLUSH : 0);
        if (rc == 0) {
                struct ost_id *oi = lsm ? &lsm->lsm_oi : &obdo->o_oi;

                obdo_refresh_inode(inode, obdo, obdo->o_valid);
                CDEBUG(D_INODE, "objid " DOSTID " size %llu, blocks %llu, blksize %lu\n",
                       POSTID(oi), i_size_read(inode),
                       (unsigned long long)inode->i_blocks,
                       1UL << inode->i_blkbits);
        }
        ccc_inode_lsm_put(inode, lsm);
        return rc;
}

int ll_merge_attr(const struct lu_env *env, struct inode *inode)
{
        struct ll_inode_info *lli = ll_i2info(inode);
        struct cl_object *obj = lli->lli_clob;
        struct cl_attr *attr = vvp_env_thread_attr(env);
        s64 atime;
        s64 mtime;
        s64 ctime;
        int rc = 0;

        ll_inode_size_lock(inode);

        /* merge timestamps the most recently obtained from mds with
         * timestamps obtained from osts
         */
        LTIME_S(inode->i_atime) = lli->lli_atime;
        LTIME_S(inode->i_mtime) = lli->lli_mtime;
        LTIME_S(inode->i_ctime) = lli->lli_ctime;

        mtime = LTIME_S(inode->i_mtime);
        atime = LTIME_S(inode->i_atime);
        ctime = LTIME_S(inode->i_ctime);

        cl_object_attr_lock(obj);
        rc = cl_object_attr_get(env, obj, attr);
        cl_object_attr_unlock(obj);

        if (rc != 0)
                goto out_size_unlock;

        if (atime < attr->cat_atime)
                atime = attr->cat_atime;

        if (ctime < attr->cat_ctime)
                ctime = attr->cat_ctime;

        if (mtime < attr->cat_mtime)
                mtime = attr->cat_mtime;

        CDEBUG(D_VFSTRACE, DFID " updating i_size %llu\n",
               PFID(&lli->lli_fid), attr->cat_size);

        i_size_write(inode, attr->cat_size);

        inode->i_blocks = attr->cat_blocks;

        LTIME_S(inode->i_mtime) = mtime;
        LTIME_S(inode->i_atime) = atime;
        LTIME_S(inode->i_ctime) = ctime;

out_size_unlock:
        ll_inode_size_unlock(inode);

        return rc;
}

int ll_glimpse_ioctl(struct ll_sb_info *sbi, struct lov_stripe_md *lsm,
                     lstat_t *st)
{
        struct obdo obdo = { 0 };
        int rc;

        rc = ll_lsm_getattr(lsm, sbi->ll_dt_exp, &obdo, 0, 0);
        if (rc == 0) {
                st->st_size   = obdo.o_size;
                st->st_blocks = obdo.o_blocks;
                st->st_mtime  = obdo.o_mtime;
                st->st_atime  = obdo.o_atime;
                st->st_ctime  = obdo.o_ctime;
        }
        return rc;
}

static bool file_is_noatime(const struct file *file)
{
        const struct vfsmount *mnt = file->f_path.mnt;
        const struct inode *inode = file_inode(file);

        /* Adapted from file_accessed() and touch_atime().*/
        if (file->f_flags & O_NOATIME)
                return true;

        if (inode->i_flags & S_NOATIME)
                return true;

        if (IS_NOATIME(inode))
                return true;

        if (mnt->mnt_flags & (MNT_NOATIME | MNT_READONLY))
                return true;

        if ((mnt->mnt_flags & MNT_NODIRATIME) && S_ISDIR(inode->i_mode))
                return true;

        if ((inode->i_sb->s_flags & MS_NODIRATIME) && S_ISDIR(inode->i_mode))
                return true;

        return false;
}

void ll_io_init(struct cl_io *io, const struct file *file, int write)
{
        struct inode *inode = file_inode(file);

        io->u.ci_rw.crw_nonblock = file->f_flags & O_NONBLOCK;
        if (write) {
                io->u.ci_wr.wr_append = !!(file->f_flags & O_APPEND);
                io->u.ci_wr.wr_sync = file->f_flags & O_SYNC ||
                                      file->f_flags & O_DIRECT ||
                                      IS_SYNC(inode);
        }
        io->ci_obj     = ll_i2info(inode)->lli_clob;
        io->ci_lockreq = CILR_MAYBE;
        if (ll_file_nolock(file)) {
                io->ci_lockreq = CILR_NEVER;
                io->ci_no_srvlock = 1;
        } else if (file->f_flags & O_APPEND) {
                io->ci_lockreq = CILR_MANDATORY;
        }

        io->ci_noatime = file_is_noatime(file);
}

static ssize_t
ll_file_io_generic(const struct lu_env *env, struct vvp_io_args *args,
                   struct file *file, enum cl_io_type iot,
                   loff_t *ppos, size_t count)
{
        struct ll_inode_info *lli = ll_i2info(file_inode(file));
        struct ll_file_data  *fd  = LUSTRE_FPRIVATE(file);
        struct cl_io     *io;
        ssize_t        result;

        CDEBUG(D_VFSTRACE, "file: %pD, type: %d ppos: %llu, count: %zd\n",
               file, iot, *ppos, count);

restart:
        io = vvp_env_thread_io(env);
        ll_io_init(io, file, iot == CIT_WRITE);

        if (cl_io_rw_init(env, io, iot, *ppos, count) == 0) {
                struct vvp_io *vio = vvp_env_io(env);
                int write_mutex_locked = 0;

                vio->vui_fd  = LUSTRE_FPRIVATE(file);
                vio->vui_io_subtype = args->via_io_subtype;

                switch (vio->vui_io_subtype) {
                case IO_NORMAL:
                        vio->vui_iter = args->u.normal.via_iter;
                        vio->vui_iocb = args->u.normal.via_iocb;
                        if ((iot == CIT_WRITE) &&
                            !(vio->vui_fd->fd_flags & LL_FILE_GROUP_LOCKED)) {
                                if (mutex_lock_interruptible(&lli->
                                                               lli_write_mutex)) {
                                        result = -ERESTARTSYS;
                                        goto out;
                                }
                                write_mutex_locked = 1;
                        }
                        down_read(&lli->lli_trunc_sem);
                        break;
                case IO_SPLICE:
                        vio->u.splice.vui_pipe = args->u.splice.via_pipe;
                        vio->u.splice.vui_flags = args->u.splice.via_flags;
                        break;
                default:
                        CERROR("Unknown IO type - %u\n", vio->vui_io_subtype);
                        LBUG();
                }
                ll_cl_add(file, env, io);
                result = cl_io_loop(env, io);
                ll_cl_remove(file, env);
                if (args->via_io_subtype == IO_NORMAL)
                        up_read(&lli->lli_trunc_sem);
                if (write_mutex_locked)
                        mutex_unlock(&lli->lli_write_mutex);
        } else {
                /* cl_io_rw_init() handled IO */
                result = io->ci_result;
        }

        if (io->ci_nob > 0) {
                result = io->ci_nob;
                *ppos = io->u.ci_wr.wr.crw_pos;
        }
        goto out;
out:
        cl_io_fini(env, io);
        /* If any bit been read/written (result != 0), we just return
         * short read/write instead of restart io.
         */
        if ((result == 0 || result == -ENODATA) && io->ci_need_restart) {
                CDEBUG(D_VFSTRACE, "Restart %s on %pD from %lld, count:%zd\n",
                       iot == CIT_READ ? "read" : "write",
                       file, *ppos, count);
                LASSERTF(io->ci_nob == 0, "%zd\n", io->ci_nob);
                goto restart;
        }

        if (iot == CIT_READ) {
                if (result >= 0)
                        ll_stats_ops_tally(ll_i2sbi(file_inode(file)),
                                           LPROC_LL_READ_BYTES, result);
        } else if (iot == CIT_WRITE) {
                if (result >= 0) {
                        ll_stats_ops_tally(ll_i2sbi(file_inode(file)),
                                           LPROC_LL_WRITE_BYTES, result);
                        fd->fd_write_failed = false;
                } else if (result != -ERESTARTSYS) {
                        fd->fd_write_failed = true;
                }
        }
        CDEBUG(D_VFSTRACE, "iot: %d, result: %zd\n", iot, result);

        return result;
}

static ssize_t ll_file_read_iter(struct kiocb *iocb, struct iov_iter *to)
{
        struct lu_env      *env;
        struct vvp_io_args *args;
        ssize_t      result;
        int              refcheck;

        env = cl_env_get(&refcheck);
        if (IS_ERR(env))
                return PTR_ERR(env);

        args = ll_env_args(env, IO_NORMAL);
        args->u.normal.via_iter = to;
        args->u.normal.via_iocb = iocb;

        result = ll_file_io_generic(env, args, iocb->ki_filp, CIT_READ,
                                    &iocb->ki_pos, iov_iter_count(to));
        cl_env_put(env, &refcheck);
        return result;
}

/*
 * Write to a file (through the page cache).
 */
static ssize_t ll_file_write_iter(struct kiocb *iocb, struct iov_iter *from)
{
        struct lu_env      *env;
        struct vvp_io_args *args;
        ssize_t      result;
        int              refcheck;

        env = cl_env_get(&refcheck);
        if (IS_ERR(env))
                return PTR_ERR(env);

        args = ll_env_args(env, IO_NORMAL);
        args->u.normal.via_iter = from;
        args->u.normal.via_iocb = iocb;

        result = ll_file_io_generic(env, args, iocb->ki_filp, CIT_WRITE,
                                    &iocb->ki_pos, iov_iter_count(from));
        cl_env_put(env, &refcheck);
        return result;
}

/*
 * Send file content (through pagecache) somewhere with helper
 */
static ssize_t ll_file_splice_read(struct file *in_file, loff_t *ppos,
                                   struct pipe_inode_info *pipe, size_t count,
                                   unsigned int flags)
{
        struct lu_env      *env;
        struct vvp_io_args *args;
        ssize_t      result;
        int              refcheck;

        env = cl_env_get(&refcheck);
        if (IS_ERR(env))
                return PTR_ERR(env);

        args = ll_env_args(env, IO_SPLICE);
        args->u.splice.via_pipe = pipe;
        args->u.splice.via_flags = flags;

        result = ll_file_io_generic(env, args, in_file, CIT_READ, ppos, count);
        cl_env_put(env, &refcheck);
        return result;
}

static int ll_lov_recreate(struct inode *inode, struct ost_id *oi, u32 ost_idx)
{
        struct obd_export *exp = ll_i2dtexp(inode);
        struct obd_trans_info oti = { 0 };
        struct obdo *oa = NULL;
        int lsm_size;
        int rc = 0;
        struct lov_stripe_md *lsm = NULL, *lsm2;

        oa = kmem_cache_zalloc(obdo_cachep, GFP_NOFS);
        if (!oa)
                return -ENOMEM;

        lsm = ccc_inode_lsm_get(inode);
        if (!lsm_has_objects(lsm)) {
                rc = -ENOENT;
                goto out;
        }

        lsm_size = sizeof(*lsm) + (sizeof(struct lov_oinfo) *
                   (lsm->lsm_stripe_count));

        lsm2 = libcfs_kvzalloc(lsm_size, GFP_NOFS);
        if (!lsm2) {
                rc = -ENOMEM;
                goto out;
        }

        oa->o_oi = *oi;
        oa->o_nlink = ost_idx;
        oa->o_flags |= OBD_FL_RECREATE_OBJS;
        oa->o_valid = OBD_MD_FLID | OBD_MD_FLFLAGS | OBD_MD_FLGROUP;
        obdo_from_inode(oa, inode, OBD_MD_FLTYPE | OBD_MD_FLATIME |
                                   OBD_MD_FLMTIME | OBD_MD_FLCTIME);
        obdo_set_parent_fid(oa, &ll_i2info(inode)->lli_fid);
        memcpy(lsm2, lsm, lsm_size);
        ll_inode_size_lock(inode);
        rc = obd_create(NULL, exp, oa, &lsm2, &oti);
        ll_inode_size_unlock(inode);

        kvfree(lsm2);
        goto out;
out:
        ccc_inode_lsm_put(inode, lsm);
        kmem_cache_free(obdo_cachep, oa);
        return rc;
}

static int ll_lov_recreate_obj(struct inode *inode, unsigned long arg)
{
        struct ll_recreate_obj ucreat;
        struct ost_id           oi;

        if (!capable(CFS_CAP_SYS_ADMIN))
                return -EPERM;

        if (copy_from_user(&ucreat, (struct ll_recreate_obj __user *)arg,
                           sizeof(ucreat)))
                return -EFAULT;

        ostid_set_seq_mdt0(&oi);
        ostid_set_id(&oi, ucreat.lrc_id);
        return ll_lov_recreate(inode, &oi, ucreat.lrc_ost_idx);
}

static int ll_lov_recreate_fid(struct inode *inode, unsigned long arg)
{
        struct lu_fid   fid;
        struct ost_id   oi;
        u32             ost_idx;

        if (!capable(CFS_CAP_SYS_ADMIN))
                return -EPERM;

        if (copy_from_user(&fid, (struct lu_fid __user *)arg, sizeof(fid)))
                return -EFAULT;

        fid_to_ostid(&fid, &oi);
        ost_idx = (fid_seq(&fid) >> 16) & 0xffff;
        return ll_lov_recreate(inode, &oi, ost_idx);
}

int ll_lov_setstripe_ea_info(struct inode *inode, struct dentry *dentry,
                             __u64 flags, struct lov_user_md *lum,
                             int lum_size)
{
        struct lov_stripe_md *lsm = NULL;
        struct lookup_intent oit = {.it_op = IT_OPEN, .it_flags = flags};
        int rc = 0;

        lsm = ccc_inode_lsm_get(inode);
        if (lsm) {
                ccc_inode_lsm_put(inode, lsm);
                CDEBUG(D_IOCTL, "stripe already exists for inode "DFID"\n",
                       PFID(ll_inode2fid(inode)));
                rc = -EEXIST;
                goto out;
        }

        ll_inode_size_lock(inode);
        oit.it_flags |= MDS_OPEN_BY_FID;
        rc = ll_intent_file_open(dentry, lum, lum_size, &oit);
        if (rc)
                goto out_unlock;
        rc = oit.it_status;
        if (rc < 0)
                goto out_req_free;

        ll_release_openhandle(inode, &oit);

out_unlock:
        ll_inode_size_unlock(inode);
        ll_intent_release(&oit);
        ccc_inode_lsm_put(inode, lsm);
out:
        return rc;
out_req_free:
        ptlrpc_req_finished((struct ptlrpc_request *)oit.it_request);
        goto out;
}

int ll_lov_getstripe_ea_info(struct inode *inode, const char *filename,
                             struct lov_mds_md **lmmp, int *lmm_size,
                             struct ptlrpc_request **request)
{
        struct ll_sb_info *sbi = ll_i2sbi(inode);
        struct mdt_body  *body;
        struct lov_mds_md *lmm = NULL;
        struct ptlrpc_request *req = NULL;
        struct md_op_data *op_data;
        int rc, lmmsize;

        rc = ll_get_default_mdsize(sbi, &lmmsize);
        if (rc)
                return rc;

        op_data = ll_prep_md_op_data(NULL, inode, NULL, filename,
                                     strlen(filename), lmmsize,
                                     LUSTRE_OPC_ANY, NULL);
        if (IS_ERR(op_data))
                return PTR_ERR(op_data);

        op_data->op_valid = OBD_MD_FLEASIZE | OBD_MD_FLDIREA;
        rc = md_getattr_name(sbi->ll_md_exp, op_data, &req);
        ll_finish_md_op_data(op_data);
        if (rc < 0) {
                CDEBUG(D_INFO, "md_getattr_name failed on %s: rc %d\n",
                       filename, rc);
                goto out;
        }

        body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);

        lmmsize = body->mbo_eadatasize;

        if (!(body->mbo_valid & (OBD_MD_FLEASIZE | OBD_MD_FLDIREA)) ||
            lmmsize == 0) {
                rc = -ENODATA;
                goto out;
        }

        lmm = req_capsule_server_sized_get(&req->rq_pill, &RMF_MDT_MD, lmmsize);

        if ((lmm->lmm_magic != cpu_to_le32(LOV_MAGIC_V1)) &&
            (lmm->lmm_magic != cpu_to_le32(LOV_MAGIC_V3))) {
                rc = -EPROTO;
                goto out;
        }

        /*
         * This is coming from the MDS, so is probably in
         * little endian.  We convert it to host endian before
         * passing it to userspace.
         */
        if (cpu_to_le32(LOV_MAGIC) != LOV_MAGIC) {
                int stripe_count;

                stripe_count = le16_to_cpu(lmm->lmm_stripe_count);
                if (le32_to_cpu(lmm->lmm_pattern) & LOV_PATTERN_F_RELEASED)
                        stripe_count = 0;

                /* if function called for directory - we should
                 * avoid swab not existent lsm objects
                 */
                if (lmm->lmm_magic == cpu_to_le32(LOV_MAGIC_V1)) {
                        lustre_swab_lov_user_md_v1((struct lov_user_md_v1 *)lmm);
                        if (S_ISREG(body->mbo_mode))
                                lustre_swab_lov_user_md_objects(
                                 ((struct lov_user_md_v1 *)lmm)->lmm_objects,
                                 stripe_count);
                } else if (lmm->lmm_magic == cpu_to_le32(LOV_MAGIC_V3)) {
                        lustre_swab_lov_user_md_v3((struct lov_user_md_v3 *)lmm);
                        if (S_ISREG(body->mbo_mode))
                                lustre_swab_lov_user_md_objects(
                                 ((struct lov_user_md_v3 *)lmm)->lmm_objects,
                                 stripe_count);
                }
        }

out:
        *lmmp = lmm;
        *lmm_size = lmmsize;
        *request = req;
        return rc;
}

static int ll_lov_setea(struct inode *inode, struct file *file,
                        unsigned long arg)
{
        __u64                    flags = MDS_OPEN_HAS_OBJS | FMODE_WRITE;
        struct lov_user_md      *lump;
        int                      lum_size = sizeof(struct lov_user_md) +
                                            sizeof(struct lov_user_ost_data);
        int                      rc;

        if (!capable(CFS_CAP_SYS_ADMIN))
                return -EPERM;

        lump = libcfs_kvzalloc(lum_size, GFP_NOFS);
        if (!lump)
                return -ENOMEM;

        if (copy_from_user(lump, (struct lov_user_md __user *)arg, lum_size)) {
                kvfree(lump);
                return -EFAULT;
        }

        rc = ll_lov_setstripe_ea_info(inode, file->f_path.dentry, flags, lump,
                                      lum_size);
        cl_lov_delay_create_clear(&file->f_flags);

        kvfree(lump);
        return rc;
}

static int ll_lov_setstripe(struct inode *inode, struct file *file,
                            unsigned long arg)
{
        struct lov_user_md_v3 lumv3;
        struct lov_user_md_v1 *lumv1 = (struct lov_user_md_v1 *)&lumv3;
        struct lov_user_md_v1 __user *lumv1p = (void __user *)arg;
        struct lov_user_md_v3 __user *lumv3p = (void __user *)arg;
        int lum_size, rc;
        __u64 flags = FMODE_WRITE;

        /* first try with v1 which is smaller than v3 */
        lum_size = sizeof(struct lov_user_md_v1);
        if (copy_from_user(lumv1, lumv1p, lum_size))
                return -EFAULT;

        if (lumv1->lmm_magic == LOV_USER_MAGIC_V3) {
                lum_size = sizeof(struct lov_user_md_v3);
                if (copy_from_user(&lumv3, lumv3p, lum_size))
                        return -EFAULT;
        }

        rc = ll_lov_setstripe_ea_info(inode, file->f_path.dentry, flags, lumv1,
                                      lum_size);
        cl_lov_delay_create_clear(&file->f_flags);
        if (rc == 0) {
                struct lov_stripe_md *lsm;
                __u32 gen;

                put_user(0, &lumv1p->lmm_stripe_count);

                ll_layout_refresh(inode, &gen);
                lsm = ccc_inode_lsm_get(inode);
                rc = obd_iocontrol(LL_IOC_LOV_GETSTRIPE, ll_i2dtexp(inode),
                                   0, lsm, (void __user *)arg);
                ccc_inode_lsm_put(inode, lsm);
        }
        return rc;
}

static int ll_lov_getstripe(struct inode *inode, unsigned long arg)
{
        struct lov_stripe_md *lsm;
        int rc = -ENODATA;

        lsm = ccc_inode_lsm_get(inode);
        if (lsm)
                rc = obd_iocontrol(LL_IOC_LOV_GETSTRIPE, ll_i2dtexp(inode), 0,
                                   lsm, (void __user *)arg);
        ccc_inode_lsm_put(inode, lsm);
        return rc;
}

static int
ll_get_grouplock(struct inode *inode, struct file *file, unsigned long arg)
{
        struct ll_inode_info   *lli = ll_i2info(inode);
        struct ll_file_data    *fd = LUSTRE_FPRIVATE(file);
        struct ll_grouplock    grouplock;
        int                  rc;

        if (arg == 0) {
                CWARN("group id for group lock must not be 0\n");
                return -EINVAL;
        }

        if (ll_file_nolock(file))
                return -EOPNOTSUPP;

        spin_lock(&lli->lli_lock);
        if (fd->fd_flags & LL_FILE_GROUP_LOCKED) {
                CWARN("group lock already existed with gid %lu\n",
                      fd->fd_grouplock.lg_gid);
                spin_unlock(&lli->lli_lock);
                return -EINVAL;
        }
        LASSERT(!fd->fd_grouplock.lg_lock);
        spin_unlock(&lli->lli_lock);

        rc = cl_get_grouplock(ll_i2info(inode)->lli_clob,
                              arg, (file->f_flags & O_NONBLOCK), &grouplock);
        if (rc)
                return rc;

        spin_lock(&lli->lli_lock);
        if (fd->fd_flags & LL_FILE_GROUP_LOCKED) {
                spin_unlock(&lli->lli_lock);
                CERROR("another thread just won the race\n");
                cl_put_grouplock(&grouplock);
                return -EINVAL;
        }

        fd->fd_flags |= LL_FILE_GROUP_LOCKED;
        fd->fd_grouplock = grouplock;
        spin_unlock(&lli->lli_lock);

        CDEBUG(D_INFO, "group lock %lu obtained\n", arg);
        return 0;
}

static int ll_put_grouplock(struct inode *inode, struct file *file,
                            unsigned long arg)
{
        struct ll_inode_info   *lli = ll_i2info(inode);
        struct ll_file_data    *fd = LUSTRE_FPRIVATE(file);
        struct ll_grouplock    grouplock;

        spin_lock(&lli->lli_lock);
        if (!(fd->fd_flags & LL_FILE_GROUP_LOCKED)) {
                spin_unlock(&lli->lli_lock);
                CWARN("no group lock held\n");
                return -EINVAL;
        }
        LASSERT(fd->fd_grouplock.lg_lock);

        if (fd->fd_grouplock.lg_gid != arg) {
                CWARN("group lock %lu doesn't match current id %lu\n",
                      arg, fd->fd_grouplock.lg_gid);
                spin_unlock(&lli->lli_lock);
                return -EINVAL;
        }

        grouplock = fd->fd_grouplock;
        memset(&fd->fd_grouplock, 0, sizeof(fd->fd_grouplock));
        fd->fd_flags &= ~LL_FILE_GROUP_LOCKED;
        spin_unlock(&lli->lli_lock);

        cl_put_grouplock(&grouplock);
        CDEBUG(D_INFO, "group lock %lu released\n", arg);
        return 0;
}

/**
 * Close inode open handle
 *
 * \param inode  [in]     inode in question
 * \param it     [in,out] intent which contains open info and result
 *
 * \retval 0     success
 * \retval <0    failure
 */
int ll_release_openhandle(struct inode *inode, struct lookup_intent *it)
{
        struct obd_client_handle *och;
        int rc;

        LASSERT(inode);

        /* Root ? Do nothing. */
        if (is_root_inode(inode))
                return 0;

        /* No open handle to close? Move away */
        if (!it_disposition(it, DISP_OPEN_OPEN))
                return 0;

        LASSERT(it_open_error(DISP_OPEN_OPEN, it) == 0);

        och = kzalloc(sizeof(*och), GFP_NOFS);
        if (!och) {
                rc = -ENOMEM;
                goto out;
        }

        ll_och_fill(ll_i2sbi(inode)->ll_md_exp, it, och);

        rc = ll_close_inode_openhandle(ll_i2sbi(inode)->ll_md_exp,
                                       inode, och, NULL);
out:
        /* this one is in place of ll_file_open */
        if (it_disposition(it, DISP_ENQ_OPEN_REF)) {
                ptlrpc_req_finished(it->it_request);
                it_clear_disposition(it, DISP_ENQ_OPEN_REF);
        }
        return rc;
}

/**
 * Get size for inode for which FIEMAP mapping is requested.
 * Make the FIEMAP get_info call and returns the result.
 */
static int ll_do_fiemap(struct inode *inode, struct ll_user_fiemap *fiemap,
                        size_t num_bytes)
{
        struct obd_export *exp = ll_i2dtexp(inode);
        struct lov_stripe_md *lsm = NULL;
        struct ll_fiemap_info_key fm_key = { .name = KEY_FIEMAP, };
        __u32 vallen = num_bytes;
        int rc;

        /* Checks for fiemap flags */
        if (fiemap->fm_flags & ~LUSTRE_FIEMAP_FLAGS_COMPAT) {
                fiemap->fm_flags &= ~LUSTRE_FIEMAP_FLAGS_COMPAT;
                return -EBADR;
        }

        /* Check for FIEMAP_FLAG_SYNC */
        if (fiemap->fm_flags & FIEMAP_FLAG_SYNC) {
                rc = filemap_fdatawrite(inode->i_mapping);
                if (rc)
                        return rc;
        }

        lsm = ccc_inode_lsm_get(inode);
        if (!lsm)
                return -ENOENT;

        /* If the stripe_count > 1 and the application does not understand
         * DEVICE_ORDER flag, then it cannot interpret the extents correctly.
         */
        if (lsm->lsm_stripe_count > 1 &&
            !(fiemap->fm_flags & FIEMAP_FLAG_DEVICE_ORDER)) {
                rc = -EOPNOTSUPP;
                goto out;
        }

        fm_key.oa.o_oi = lsm->lsm_oi;
        fm_key.oa.o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;

        if (i_size_read(inode) == 0) {
                rc = ll_glimpse_size(inode);
                if (rc)
                        goto out;
        }

        obdo_from_inode(&fm_key.oa, inode, OBD_MD_FLSIZE);
        obdo_set_parent_fid(&fm_key.oa, &ll_i2info(inode)->lli_fid);
        /* If filesize is 0, then there would be no objects for mapping */
        if (fm_key.oa.o_size == 0) {
                fiemap->fm_mapped_extents = 0;
                rc = 0;
                goto out;
        }

        memcpy(&fm_key.fiemap, fiemap, sizeof(*fiemap));

        rc = obd_get_info(NULL, exp, sizeof(fm_key), &fm_key, &vallen,
                          fiemap, lsm);
        if (rc)
                CERROR("obd_get_info failed: rc = %d\n", rc);

out:
        ccc_inode_lsm_put(inode, lsm);
        return rc;
}

int ll_fid2path(struct inode *inode, void __user *arg)
{
        struct obd_export *exp = ll_i2mdexp(inode);
        const struct getinfo_fid2path __user *gfin = arg;
        struct getinfo_fid2path *gfout;
        u32 pathlen;
        size_t outsize;
        int rc;

        if (!capable(CFS_CAP_DAC_READ_SEARCH) &&
            !(ll_i2sbi(inode)->ll_flags & LL_SBI_USER_FID2PATH))
                return -EPERM;

        /* Only need to get the buflen */
        if (get_user(pathlen, &gfin->gf_pathlen))
                return -EFAULT;

        if (pathlen > PATH_MAX)
                return -EINVAL;

        outsize = sizeof(*gfout) + pathlen;

        gfout = kzalloc(outsize, GFP_NOFS);
        if (!gfout)
                return -ENOMEM;

        if (copy_from_user(gfout, arg, sizeof(*gfout))) {
                rc = -EFAULT;
                goto gf_free;
        }

        /* Call mdc_iocontrol */
        rc = obd_iocontrol(OBD_IOC_FID2PATH, exp, outsize, gfout, NULL);
        if (rc != 0)
                goto gf_free;

        if (copy_to_user(arg, gfout, outsize))
                rc = -EFAULT;

gf_free:
        kfree(gfout);
        return rc;
}

static int ll_ioctl_fiemap(struct inode *inode, unsigned long arg)
{
        struct ll_user_fiemap *fiemap_s;
        size_t num_bytes, ret_bytes;
        unsigned int extent_count;
        int rc = 0;

        /* Get the extent count so we can calculate the size of
         * required fiemap buffer
         */
        if (get_user(extent_count,
                     &((struct ll_user_fiemap __user *)arg)->fm_extent_count))
                return -EFAULT;

        if (extent_count >=
            (SIZE_MAX - sizeof(*fiemap_s)) / sizeof(struct ll_fiemap_extent))
                return -EINVAL;
        num_bytes = sizeof(*fiemap_s) + (extent_count *
                                         sizeof(struct ll_fiemap_extent));

        fiemap_s = libcfs_kvzalloc(num_bytes, GFP_NOFS);
        if (!fiemap_s)
                return -ENOMEM;

        /* get the fiemap value */
        if (copy_from_user(fiemap_s, (struct ll_user_fiemap __user *)arg,
                           sizeof(*fiemap_s))) {
                rc = -EFAULT;
                goto error;
        }

        /* If fm_extent_count is non-zero, read the first extent since
         * it is used to calculate end_offset and device from previous
         * fiemap call.
         */
        if (extent_count) {
                if (copy_from_user(&fiemap_s->fm_extents[0],
                                   (char __user *)arg + sizeof(*fiemap_s),
                                   sizeof(struct ll_fiemap_extent))) {
                        rc = -EFAULT;
                        goto error;
                }
        }

        rc = ll_do_fiemap(inode, fiemap_s, num_bytes);
        if (rc)
                goto error;

        ret_bytes = sizeof(struct ll_user_fiemap);

        if (extent_count != 0)
                ret_bytes += (fiemap_s->fm_mapped_extents *
                                 sizeof(struct ll_fiemap_extent));

        if (copy_to_user((void __user *)arg, fiemap_s, ret_bytes))
                rc = -EFAULT;

error:
        kvfree(fiemap_s);
        return rc;
}

/*
 * Read the data_version for inode.
 *
 * This value is computed using stripe object version on OST.
 * Version is computed using server side locking.
 *
 * @param sync  if do sync on the OST side;
 *              0: no sync
 *              LL_DV_RD_FLUSH: flush dirty pages, LCK_PR on OSTs
 *              LL_DV_WR_FLUSH: drop all caching pages, LCK_PW on OSTs
 */
int ll_data_version(struct inode *inode, __u64 *data_version, int flags)
{
        struct lov_stripe_md    *lsm = NULL;
        struct ll_sb_info       *sbi = ll_i2sbi(inode);
        struct obdo             *obdo = NULL;
        int                      rc;

        /* If no stripe, we consider version is 0. */
        lsm = ccc_inode_lsm_get(inode);
        if (!lsm_has_objects(lsm)) {
                *data_version = 0;
                CDEBUG(D_INODE, "No object for inode\n");
                rc = 0;
                goto out;
        }

        obdo = kzalloc(sizeof(*obdo), GFP_NOFS);
        if (!obdo) {
                rc = -ENOMEM;
                goto out;
        }

        rc = ll_lsm_getattr(lsm, sbi->ll_dt_exp, obdo, 0, flags);
        if (rc == 0) {
                if (!(obdo->o_valid & OBD_MD_FLDATAVERSION))
                        rc = -EOPNOTSUPP;
                else
                        *data_version = obdo->o_data_version;
        }

        kfree(obdo);
out:
        ccc_inode_lsm_put(inode, lsm);
        return rc;
}

/*
 * Trigger a HSM release request for the provided inode.
 */
int ll_hsm_release(struct inode *inode)
{
        struct cl_env_nest nest;
        struct lu_env *env;
        struct obd_client_handle *och = NULL;
        __u64 data_version = 0;
        int rc;

        CDEBUG(D_INODE, "%s: Releasing file "DFID".\n",
               ll_get_fsname(inode->i_sb, NULL, 0),
               PFID(&ll_i2info(inode)->lli_fid));

        och = ll_lease_open(inode, NULL, FMODE_WRITE, MDS_OPEN_RELEASE);
        if (IS_ERR(och)) {
                rc = PTR_ERR(och);
                goto out;
        }

        /* Grab latest data_version and [am]time values */
        rc = ll_data_version(inode, &data_version, LL_DV_WR_FLUSH);
        if (rc != 0)
                goto out;

        env = cl_env_nested_get(&nest);
        if (IS_ERR(env)) {
                rc = PTR_ERR(env);
                goto out;
        }

        ll_merge_attr(env, inode);
        cl_env_nested_put(&nest, env);

        /* Release the file.
         * NB: lease lock handle is released in mdc_hsm_release_pack() because
         * we still need it to pack l_remote_handle to MDT.
         */
        rc = ll_close_inode_openhandle(ll_i2sbi(inode)->ll_md_exp, inode, och,
                                       &data_version);
        och = NULL;

out:
        if (och && !IS_ERR(och)) /* close the file */
                ll_lease_close(och, inode, NULL);

        return rc;
}

struct ll_swap_stack {
        struct iattr             ia1, ia2;
        __u64                    dv1, dv2;
        struct inode            *inode1, *inode2;
        bool                     check_dv1, check_dv2;
};

static int ll_swap_layouts(struct file *file1, struct file *file2,
                           struct lustre_swap_layouts *lsl)
{
        struct mdc_swap_layouts  msl;
        struct md_op_data       *op_data;
        __u32                    gid;
        __u64                    dv;
        struct ll_swap_stack    *llss = NULL;
        int                      rc;

        llss = kzalloc(sizeof(*llss), GFP_NOFS);
        if (!llss)
                return -ENOMEM;

        llss->inode1 = file_inode(file1);
        llss->inode2 = file_inode(file2);

        if (!S_ISREG(llss->inode2->i_mode)) {
                rc = -EINVAL;
                goto free;
        }

        if (inode_permission(llss->inode1, MAY_WRITE) ||
            inode_permission(llss->inode2, MAY_WRITE)) {
                rc = -EPERM;
                goto free;
        }

        if (llss->inode2->i_sb != llss->inode1->i_sb) {
                rc = -EXDEV;
                goto free;
        }

        /* we use 2 bool because it is easier to swap than 2 bits */
        if (lsl->sl_flags & SWAP_LAYOUTS_CHECK_DV1)
                llss->check_dv1 = true;

        if (lsl->sl_flags & SWAP_LAYOUTS_CHECK_DV2)
                llss->check_dv2 = true;

        /* we cannot use lsl->sl_dvX directly because we may swap them */
        llss->dv1 = lsl->sl_dv1;
        llss->dv2 = lsl->sl_dv2;

        rc = lu_fid_cmp(ll_inode2fid(llss->inode1), ll_inode2fid(llss->inode2));
        if (rc == 0) /* same file, done! */ {
                rc = 0;
                goto free;
        }

        if (rc < 0) { /* sequentialize it */
                swap(llss->inode1, llss->inode2);
                swap(file1, file2);
                swap(llss->dv1, llss->dv2);
                swap(llss->check_dv1, llss->check_dv2);
        }

        gid = lsl->sl_gid;
        if (gid != 0) { /* application asks to flush dirty cache */
                rc = ll_get_grouplock(llss->inode1, file1, gid);
                if (rc < 0)
                        goto free;

                rc = ll_get_grouplock(llss->inode2, file2, gid);
                if (rc < 0) {
                        ll_put_grouplock(llss->inode1, file1, gid);
                        goto free;
                }
        }

        /* to be able to restore mtime and atime after swap
         * we need to first save them
         */
        if (lsl->sl_flags &
            (SWAP_LAYOUTS_KEEP_MTIME | SWAP_LAYOUTS_KEEP_ATIME)) {
                llss->ia1.ia_mtime = llss->inode1->i_mtime;
                llss->ia1.ia_atime = llss->inode1->i_atime;
                llss->ia1.ia_valid = ATTR_MTIME | ATTR_ATIME;
                llss->ia2.ia_mtime = llss->inode2->i_mtime;
                llss->ia2.ia_atime = llss->inode2->i_atime;
                llss->ia2.ia_valid = ATTR_MTIME | ATTR_ATIME;
        }

        /* ultimate check, before swapping the layouts we check if
         * dataversion has changed (if requested)
         */
        if (llss->check_dv1) {
                rc = ll_data_version(llss->inode1, &dv, 0);
                if (rc)
                        goto putgl;
                if (dv != llss->dv1) {
                        rc = -EAGAIN;
                        goto putgl;
                }
        }

        if (llss->check_dv2) {
                rc = ll_data_version(llss->inode2, &dv, 0);
                if (rc)
                        goto putgl;
                if (dv != llss->dv2) {
                        rc = -EAGAIN;
                        goto putgl;
                }
        }

        /* struct md_op_data is used to send the swap args to the mdt
         * only flags is missing, so we use struct mdc_swap_layouts
         * through the md_op_data->op_data
         */
        /* flags from user space have to be converted before they are send to
         * server, no flag is sent today, they are only used on the client
         */
        msl.msl_flags = 0;
        rc = -ENOMEM;
        op_data = ll_prep_md_op_data(NULL, llss->inode1, llss->inode2, NULL, 0,
                                     0, LUSTRE_OPC_ANY, &msl);
        if (IS_ERR(op_data)) {
                rc = PTR_ERR(op_data);
                goto free;
        }

        rc = obd_iocontrol(LL_IOC_LOV_SWAP_LAYOUTS, ll_i2mdexp(llss->inode1),
                           sizeof(*op_data), op_data, NULL);
        ll_finish_md_op_data(op_data);

putgl:
        if (gid != 0) {
                ll_put_grouplock(llss->inode2, file2, gid);
                ll_put_grouplock(llss->inode1, file1, gid);
        }

        /* rc can be set from obd_iocontrol() or from a GOTO(putgl, ...) */
        if (rc != 0)
                goto free;

        /* clear useless flags */
        if (!(lsl->sl_flags & SWAP_LAYOUTS_KEEP_MTIME)) {
                llss->ia1.ia_valid &= ~ATTR_MTIME;
                llss->ia2.ia_valid &= ~ATTR_MTIME;
        }

        if (!(lsl->sl_flags & SWAP_LAYOUTS_KEEP_ATIME)) {
                llss->ia1.ia_valid &= ~ATTR_ATIME;
                llss->ia2.ia_valid &= ~ATTR_ATIME;
        }

        /* update time if requested */
        rc = 0;
        if (llss->ia2.ia_valid != 0) {
                inode_lock(llss->inode1);
                rc = ll_setattr(file1->f_path.dentry, &llss->ia2);
                inode_unlock(llss->inode1);
        }

        if (llss->ia1.ia_valid != 0) {
                int rc1;

                inode_lock(llss->inode2);
                rc1 = ll_setattr(file2->f_path.dentry, &llss->ia1);
                inode_unlock(llss->inode2);
                if (rc == 0)
                        rc = rc1;
        }

free:
        kfree(llss);

        return rc;
}

static int ll_hsm_state_set(struct inode *inode, struct hsm_state_set *hss)
{
        struct md_op_data       *op_data;
        int                      rc;

        /* Detect out-of range masks */
        if ((hss->hss_setmask | hss->hss_clearmask) & ~HSM_FLAGS_MASK)
                return -EINVAL;

        /* Non-root users are forbidden to set or clear flags which are
         * NOT defined in HSM_USER_MASK.
         */
        if (((hss->hss_setmask | hss->hss_clearmask) & ~HSM_USER_MASK) &&
            !capable(CFS_CAP_SYS_ADMIN))
                return -EPERM;

        /* Detect out-of range archive id */
        if ((hss->hss_valid & HSS_ARCHIVE_ID) &&
            (hss->hss_archive_id > LL_HSM_MAX_ARCHIVE))
                return -EINVAL;

        op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL, 0, 0,
                                     LUSTRE_OPC_ANY, hss);
        if (IS_ERR(op_data))
                return PTR_ERR(op_data);

        rc = obd_iocontrol(LL_IOC_HSM_STATE_SET, ll_i2mdexp(inode),
                           sizeof(*op_data), op_data, NULL);

        ll_finish_md_op_data(op_data);

        return rc;
}

static int ll_hsm_import(struct inode *inode, struct file *file,
                         struct hsm_user_import *hui)
{
        struct hsm_state_set    *hss = NULL;
        struct iattr            *attr = NULL;
        int                      rc;

        if (!S_ISREG(inode->i_mode))
                return -EINVAL;

        /* set HSM flags */
        hss = kzalloc(sizeof(*hss), GFP_NOFS);
        if (!hss)
                return -ENOMEM;

        hss->hss_valid = HSS_SETMASK | HSS_ARCHIVE_ID;
        hss->hss_archive_id = hui->hui_archive_id;
        hss->hss_setmask = HS_ARCHIVED | HS_EXISTS | HS_RELEASED;
        rc = ll_hsm_state_set(inode, hss);
        if (rc != 0)
                goto free_hss;

        attr = kzalloc(sizeof(*attr), GFP_NOFS);
        if (!attr) {
                rc = -ENOMEM;
                goto free_hss;
        }

        attr->ia_mode = hui->hui_mode & (S_IRWXU | S_IRWXG | S_IRWXO);
        attr->ia_mode |= S_IFREG;
        attr->ia_uid = make_kuid(&init_user_ns, hui->hui_uid);
        attr->ia_gid = make_kgid(&init_user_ns, hui->hui_gid);
        attr->ia_size = hui->hui_size;
        attr->ia_mtime.tv_sec = hui->hui_mtime;
        attr->ia_mtime.tv_nsec = hui->hui_mtime_ns;
        attr->ia_atime.tv_sec = hui->hui_atime;
        attr->ia_atime.tv_nsec = hui->hui_atime_ns;

        attr->ia_valid = ATTR_SIZE | ATTR_MODE | ATTR_FORCE |
                         ATTR_UID | ATTR_GID |
                         ATTR_MTIME | ATTR_MTIME_SET |
                         ATTR_ATIME | ATTR_ATIME_SET;

        inode_lock(inode);

        rc = ll_setattr_raw(file->f_path.dentry, attr, true);
        if (rc == -ENODATA)
                rc = 0;

        inode_unlock(inode);

        kfree(attr);
free_hss:
        kfree(hss);
        return rc;
}

static long
ll_file_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
{
        struct inode            *inode = file_inode(file);
        struct ll_file_data     *fd = LUSTRE_FPRIVATE(file);
        int                      flags, rc;

        CDEBUG(D_VFSTRACE, "VFS Op:inode="DFID"(%p),cmd=%x\n",
               PFID(ll_inode2fid(inode)), inode, cmd);
        ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_IOCTL, 1);

        /* asm-ppc{,64} declares TCGETS, et. al. as type 't' not 'T' */
        if (_IOC_TYPE(cmd) == 'T' || _IOC_TYPE(cmd) == 't') /* tty ioctls */
                return -ENOTTY;

        switch (cmd) {
        case LL_IOC_GETFLAGS:
                /* Get the current value of the file flags */
                return put_user(fd->fd_flags, (int __user *)arg);
        case LL_IOC_SETFLAGS:
        case LL_IOC_CLRFLAGS:
                /* Set or clear specific file flags */
                /* XXX This probably needs checks to ensure the flags are
                 *     not abused, and to handle any flag side effects.
                 */
                if (get_user(flags, (int __user *)arg))
                        return -EFAULT;

                if (cmd == LL_IOC_SETFLAGS) {
                        if ((flags & LL_FILE_IGNORE_LOCK) &&
                            !(file->f_flags & O_DIRECT)) {
                                CERROR("%s: unable to disable locking on non-O_DIRECT file\n",
                                       current->comm);
                                return -EINVAL;
                        }

                        fd->fd_flags |= flags;
                } else {
                        fd->fd_flags &= ~flags;
                }
                return 0;
        case LL_IOC_LOV_SETSTRIPE:
                return ll_lov_setstripe(inode, file, arg);
        case LL_IOC_LOV_SETEA:
                return ll_lov_setea(inode, file, arg);
        case LL_IOC_LOV_SWAP_LAYOUTS: {
                struct file *file2;
                struct lustre_swap_layouts lsl;

                if (copy_from_user(&lsl, (char __user *)arg,
                                   sizeof(struct lustre_swap_layouts)))
                        return -EFAULT;

                if ((file->f_flags & O_ACCMODE) == 0) /* O_RDONLY */
                        return -EPERM;

                file2 = fget(lsl.sl_fd);
                if (!file2)
                        return -EBADF;

                rc = -EPERM;
                if ((file2->f_flags & O_ACCMODE) != 0) /* O_WRONLY or O_RDWR */
                        rc = ll_swap_layouts(file, file2, &lsl);
                fput(file2);
                return rc;
        }
        case LL_IOC_LOV_GETSTRIPE:
                return ll_lov_getstripe(inode, arg);
        case LL_IOC_RECREATE_OBJ:
                return ll_lov_recreate_obj(inode, arg);
        case LL_IOC_RECREATE_FID:
                return ll_lov_recreate_fid(inode, arg);
        case FSFILT_IOC_FIEMAP:
                return ll_ioctl_fiemap(inode, arg);
        case FSFILT_IOC_GETFLAGS:
        case FSFILT_IOC_SETFLAGS:
                return ll_iocontrol(inode, file, cmd, arg);
        case FSFILT_IOC_GETVERSION_OLD:
        case FSFILT_IOC_GETVERSION:
                return put_user(inode->i_generation, (int __user *)arg);
        case LL_IOC_GROUP_LOCK:
                return ll_get_grouplock(inode, file, arg);
        case LL_IOC_GROUP_UNLOCK:
                return ll_put_grouplock(inode, file, arg);
        case IOC_OBD_STATFS:
                return ll_obd_statfs(inode, (void __user *)arg);

        /* We need to special case any other ioctls we want to handle,
         * to send them to the MDS/OST as appropriate and to properly
         * network encode the arg field.
        case FSFILT_IOC_SETVERSION_OLD:
        case FSFILT_IOC_SETVERSION:
        */
        case LL_IOC_FLUSHCTX:
                return ll_flush_ctx(inode);
        case LL_IOC_PATH2FID: {
                if (copy_to_user((void __user *)arg, ll_inode2fid(inode),
                                 sizeof(struct lu_fid)))
                        return -EFAULT;

                return 0;
        }
        case OBD_IOC_FID2PATH:
                return ll_fid2path(inode, (void __user *)arg);
        case LL_IOC_DATA_VERSION: {
                struct ioc_data_version idv;
                int                     rc;

                if (copy_from_user(&idv, (char __user *)arg, sizeof(idv)))
                        return -EFAULT;

                idv.idv_flags &= LL_DV_RD_FLUSH | LL_DV_WR_FLUSH;
                rc = ll_data_version(inode, &idv.idv_version, idv.idv_flags);
                if (rc == 0 && copy_to_user((char __user *)arg, &idv,
                                            sizeof(idv)))
                        return -EFAULT;

                return rc;
        }

        case LL_IOC_GET_MDTIDX: {
                int mdtidx;

                mdtidx = ll_get_mdt_idx(inode);
                if (mdtidx < 0)
                        return mdtidx;

                if (put_user(mdtidx, (int __user *)arg))
                        return -EFAULT;

                return 0;
        }
        case OBD_IOC_GETDTNAME:
        case OBD_IOC_GETMDNAME:
                return ll_get_obd_name(inode, cmd, arg);
        case LL_IOC_HSM_STATE_GET: {
                struct md_op_data       *op_data;
                struct hsm_user_state   *hus;
                int                      rc;

                hus = kzalloc(sizeof(*hus), GFP_NOFS);
                if (!hus)
                        return -ENOMEM;

                op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL, 0, 0,
                                             LUSTRE_OPC_ANY, hus);
                if (IS_ERR(op_data)) {
                        kfree(hus);
                        return PTR_ERR(op_data);
                }

                rc = obd_iocontrol(cmd, ll_i2mdexp(inode), sizeof(*op_data),
                                   op_data, NULL);

                if (copy_to_user((void __user *)arg, hus, sizeof(*hus)))
                        rc = -EFAULT;

                ll_finish_md_op_data(op_data);
                kfree(hus);
                return rc;
        }
        case LL_IOC_HSM_STATE_SET: {
                struct hsm_state_set    *hss;
                int                      rc;

                hss = memdup_user((char __user *)arg, sizeof(*hss));
                if (IS_ERR(hss))
                        return PTR_ERR(hss);

                rc = ll_hsm_state_set(inode, hss);

                kfree(hss);
                return rc;
        }
        case LL_IOC_HSM_ACTION: {
                struct md_op_data               *op_data;
                struct hsm_current_action       *hca;
                int                              rc;

                hca = kzalloc(sizeof(*hca), GFP_NOFS);
                if (!hca)
                        return -ENOMEM;

                op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL, 0, 0,
                                             LUSTRE_OPC_ANY, hca);
                if (IS_ERR(op_data)) {
                        kfree(hca);
                        return PTR_ERR(op_data);
                }

                rc = obd_iocontrol(cmd, ll_i2mdexp(inode), sizeof(*op_data),
                                   op_data, NULL);

                if (copy_to_user((char __user *)arg, hca, sizeof(*hca)))
                        rc = -EFAULT;

                ll_finish_md_op_data(op_data);
                kfree(hca);
                return rc;
        }
        case LL_IOC_SET_LEASE: {
                struct ll_inode_info *lli = ll_i2info(inode);
                struct obd_client_handle *och = NULL;
                bool lease_broken;
                fmode_t mode = 0;

                switch (arg) {
                case F_WRLCK:
                        if (!(file->f_mode & FMODE_WRITE))
                                return -EPERM;
                        mode = FMODE_WRITE;
                        break;
                case F_RDLCK:
                        if (!(file->f_mode & FMODE_READ))
                                return -EPERM;
                        mode = FMODE_READ;
                        break;
                case F_UNLCK:
                        mutex_lock(&lli->lli_och_mutex);
                        if (fd->fd_lease_och) {
                                och = fd->fd_lease_och;
                                fd->fd_lease_och = NULL;
                        }
                        mutex_unlock(&lli->lli_och_mutex);

                        if (och) {
                                mode = och->och_flags &
                                       (FMODE_READ | FMODE_WRITE);
                                rc = ll_lease_close(och, inode, &lease_broken);
                                if (rc == 0 && lease_broken)
                                        mode = 0;
                        } else {
                                rc = -ENOLCK;
                        }

                        /* return the type of lease or error */
                        return rc < 0 ? rc : (int)mode;
                default:
                        return -EINVAL;
                }

                CDEBUG(D_INODE, "Set lease with mode %d\n", mode);

                /* apply for lease */
                och = ll_lease_open(inode, file, mode, 0);
                if (IS_ERR(och))
                        return PTR_ERR(och);

                rc = 0;
                mutex_lock(&lli->lli_och_mutex);
                if (!fd->fd_lease_och) {
                        fd->fd_lease_och = och;
                        och = NULL;
                }
                mutex_unlock(&lli->lli_och_mutex);
                if (och) {
                        /* impossible now that only excl is supported for now */
                        ll_lease_close(och, inode, &lease_broken);
                        rc = -EBUSY;
                }
                return rc;
        }
        case LL_IOC_GET_LEASE: {
                struct ll_inode_info *lli = ll_i2info(inode);
                struct ldlm_lock *lock = NULL;

                rc = 0;
                mutex_lock(&lli->lli_och_mutex);
                if (fd->fd_lease_och) {
                        struct obd_client_handle *och = fd->fd_lease_och;

                        lock = ldlm_handle2lock(&och->och_lease_handle);
                        if (lock) {
                                lock_res_and_lock(lock);
                                if (!ldlm_is_cancel(lock))
                                        rc = och->och_flags &
                                                (FMODE_READ | FMODE_WRITE);
                                unlock_res_and_lock(lock);
                                LDLM_LOCK_PUT(lock);
                        }
                }
                mutex_unlock(&lli->lli_och_mutex);
                return rc;
        }
        case LL_IOC_HSM_IMPORT: {
                struct hsm_user_import *hui;

                hui = memdup_user((void __user *)arg, sizeof(*hui));
                if (IS_ERR(hui))
                        return PTR_ERR(hui);

                rc = ll_hsm_import(inode, file, hui);

                kfree(hui);
                return rc;
        }
        default: {
                int err;

                if (ll_iocontrol_call(inode, file, cmd, arg, &err) ==
                     LLIOC_STOP)
                        return err;

                return obd_iocontrol(cmd, ll_i2dtexp(inode), 0, NULL,
                                     (void __user *)arg);
        }
        }
}

static loff_t ll_file_seek(struct file *file, loff_t offset, int origin)
{
        struct inode *inode = file_inode(file);
        loff_t retval, eof = 0;

        retval = offset + ((origin == SEEK_END) ? i_size_read(inode) :
                           (origin == SEEK_CUR) ? file->f_pos : 0);
        CDEBUG(D_VFSTRACE, "VFS Op:inode="DFID"(%p), to=%llu=%#llx(%d)\n",
               PFID(ll_inode2fid(inode)), inode, retval, retval, origin);
        ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_LLSEEK, 1);

        if (origin == SEEK_END || origin == SEEK_HOLE || origin == SEEK_DATA) {
                retval = ll_glimpse_size(inode);
                if (retval != 0)
                        return retval;
                eof = i_size_read(inode);
        }

        retval = generic_file_llseek_size(file, offset, origin,
                                          ll_file_maxbytes(inode), eof);
        return retval;
}

static int ll_flush(struct file *file, fl_owner_t id)
{
        struct inode *inode = file_inode(file);
        struct ll_inode_info *lli = ll_i2info(inode);
        struct ll_file_data *fd = LUSTRE_FPRIVATE(file);
        int rc, err;

        LASSERT(!S_ISDIR(inode->i_mode));

        /* catch async errors that were recorded back when async writeback
         * failed for pages in this mapping.
         */
        rc = lli->lli_async_rc;
        lli->lli_async_rc = 0;
        if (lli->lli_clob) {
                err = lov_read_and_clear_async_rc(lli->lli_clob);
                if (!rc)
                        rc = err;
        }

        /* The application has been told about write failure already.
         * Do not report failure again.
         */
        if (fd->fd_write_failed)
                return 0;
        return rc ? -EIO : 0;
}

/**
 * Called to make sure a portion of file has been written out.
 * if @mode is not CL_FSYNC_LOCAL, it will send OST_SYNC RPCs to OST.
 *
 * Return how many pages have been written.
 */
int cl_sync_file_range(struct inode *inode, loff_t start, loff_t end,
                       enum cl_fsync_mode mode, int ignore_layout)
{
        struct cl_env_nest nest;
        struct lu_env *env;
        struct cl_io *io;
        struct cl_fsync_io *fio;
        int result;

        if (mode != CL_FSYNC_NONE && mode != CL_FSYNC_LOCAL &&
            mode != CL_FSYNC_DISCARD && mode != CL_FSYNC_ALL)
                return -EINVAL;

        env = cl_env_nested_get(&nest);
        if (IS_ERR(env))
                return PTR_ERR(env);

        io = vvp_env_thread_io(env);
        io->ci_obj = ll_i2info(inode)->lli_clob;
        io->ci_ignore_layout = ignore_layout;

        /* initialize parameters for sync */
        fio = &io->u.ci_fsync;
        fio->fi_start = start;
        fio->fi_end = end;
        fio->fi_fid = ll_inode2fid(inode);
        fio->fi_mode = mode;
        fio->fi_nr_written = 0;

        if (cl_io_init(env, io, CIT_FSYNC, io->ci_obj) == 0)
                result = cl_io_loop(env, io);
        else
                result = io->ci_result;
        if (result == 0)
                result = fio->fi_nr_written;
        cl_io_fini(env, io);
        cl_env_nested_put(&nest, env);

        return result;
}

int ll_fsync(struct file *file, loff_t start, loff_t end, int datasync)
{
        struct inode *inode = file_inode(file);
        struct ll_inode_info *lli = ll_i2info(inode);
        struct ptlrpc_request *req;
        int rc, err;

        CDEBUG(D_VFSTRACE, "VFS Op:inode="DFID"(%p)\n",
               PFID(ll_inode2fid(inode)), inode);
        ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_FSYNC, 1);

        rc = filemap_write_and_wait_range(inode->i_mapping, start, end);
        inode_lock(inode);

        /* catch async errors that were recorded back when async writeback
         * failed for pages in this mapping.
         */
        if (!S_ISDIR(inode->i_mode)) {
                err = lli->lli_async_rc;
                lli->lli_async_rc = 0;
                if (rc == 0)
                        rc = err;
                err = lov_read_and_clear_async_rc(lli->lli_clob);
                if (rc == 0)
                        rc = err;
        }

        err = md_sync(ll_i2sbi(inode)->ll_md_exp, ll_inode2fid(inode), &req);
        if (!rc)
                rc = err;
        if (!err)
                ptlrpc_req_finished(req);

        if (S_ISREG(inode->i_mode)) {
                struct ll_file_data *fd = LUSTRE_FPRIVATE(file);

                err = cl_sync_file_range(inode, start, end, CL_FSYNC_ALL, 0);
                if (rc == 0 && err < 0)
                        rc = err;
                if (rc < 0)
                        fd->fd_write_failed = true;
                else
                        fd->fd_write_failed = false;
        }

        inode_unlock(inode);
        return rc;
}

static int
ll_file_flock(struct file *file, int cmd, struct file_lock *file_lock)
{
        struct inode *inode = file_inode(file);
        struct ll_sb_info *sbi = ll_i2sbi(inode);
        struct ldlm_enqueue_info einfo = {
                .ei_type        = LDLM_FLOCK,
                .ei_cb_cp       = ldlm_flock_completion_ast,
                .ei_cbdata      = file_lock,
        };
        struct md_op_data *op_data;
        struct lustre_handle lockh = {0};
        ldlm_policy_data_t flock = { {0} };
        int fl_type = file_lock->fl_type;
        __u64 flags = 0;
        int rc;
        int rc2 = 0;

        CDEBUG(D_VFSTRACE, "VFS Op:inode="DFID" file_lock=%p\n",
               PFID(ll_inode2fid(inode)), file_lock);

        ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_FLOCK, 1);

        if (file_lock->fl_flags & FL_FLOCK)
                LASSERT((cmd == F_SETLKW) || (cmd == F_SETLK));
        else if (!(file_lock->fl_flags & FL_POSIX))
                return -EINVAL;

        flock.l_flock.owner = (unsigned long)file_lock->fl_owner;
        flock.l_flock.pid = file_lock->fl_pid;
        flock.l_flock.start = file_lock->fl_start;
        flock.l_flock.end = file_lock->fl_end;

        /* Somewhat ugly workaround for svc lockd.
         * lockd installs custom fl_lmops->lm_compare_owner that checks
         * for the fl_owner to be the same (which it always is on local node
         * I guess between lockd processes) and then compares pid.
         * As such we assign pid to the owner field to make it all work,
         * conflict with normal locks is unlikely since pid space and
         * pointer space for current->files are not intersecting
         */
        if (file_lock->fl_lmops && file_lock->fl_lmops->lm_compare_owner)
                flock.l_flock.owner = (unsigned long)file_lock->fl_pid;

        switch (fl_type) {
        case F_RDLCK:
                einfo.ei_mode = LCK_PR;
                break;
        case F_UNLCK:
                /* An unlock request may or may not have any relation to
                 * existing locks so we may not be able to pass a lock handle
                 * via a normal ldlm_lock_cancel() request. The request may even
                 * unlock a byte range in the middle of an existing lock. In
                 * order to process an unlock request we need all of the same
                 * information that is given with a normal read or write record
                 * lock request. To avoid creating another ldlm unlock (cancel)
                 * message we'll treat a LCK_NL flock request as an unlock.
                 */
                einfo.ei_mode = LCK_NL;
                break;
        case F_WRLCK:
                einfo.ei_mode = LCK_PW;
                break;
        default:
                CDEBUG(D_INFO, "Unknown fcntl lock type: %d\n", fl_type);
                return -ENOTSUPP;
        }

        switch (cmd) {
        case F_SETLKW:
#ifdef F_SETLKW64
        case F_SETLKW64:
#endif
                flags = 0;
                break;
        case F_SETLK:
#ifdef F_SETLK64
        case F_SETLK64:
#endif
                flags = LDLM_FL_BLOCK_NOWAIT;
                break;
        case F_GETLK:
#ifdef F_GETLK64
        case F_GETLK64:
#endif
                flags = LDLM_FL_TEST_LOCK;
                break;
        default:
                CERROR("unknown fcntl lock command: %d\n", cmd);
                return -EINVAL;
        }

        /*
         * Save the old mode so that if the mode in the lock changes we
         * can decrement the appropriate reader or writer refcount.
         */
        file_lock->fl_type = einfo.ei_mode;

        op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL, 0, 0,
                                     LUSTRE_OPC_ANY, NULL);
        if (IS_ERR(op_data))
                return PTR_ERR(op_data);

        CDEBUG(D_DLMTRACE, "inode="DFID", pid=%u, flags=%#llx, mode=%u, start=%llu, end=%llu\n",
               PFID(ll_inode2fid(inode)), flock.l_flock.pid, flags,
               einfo.ei_mode, flock.l_flock.start, flock.l_flock.end);

        rc = md_enqueue(sbi->ll_md_exp, &einfo, &flock, NULL, op_data, &lockh,
                        flags);

        /* Restore the file lock type if not TEST lock. */
        if (!(flags & LDLM_FL_TEST_LOCK))
                file_lock->fl_type = fl_type;

        if ((rc == 0 || file_lock->fl_type == F_UNLCK) &&
            !(flags & LDLM_FL_TEST_LOCK))
                rc2  = locks_lock_file_wait(file, file_lock);

        if (rc2 && file_lock->fl_type != F_UNLCK) {
                einfo.ei_mode = LCK_NL;
                md_enqueue(sbi->ll_md_exp, &einfo, &flock, NULL, op_data,
                           &lockh, flags);
                rc = rc2;
        }

        ll_finish_md_op_data(op_data);

        return rc;
}

int ll_get_fid_by_name(struct inode *parent, const char *name,
                       int namelen, struct lu_fid *fid)
{
        struct md_op_data *op_data = NULL;
        struct ptlrpc_request *req;
        struct mdt_body *body;
        int rc;

        op_data = ll_prep_md_op_data(NULL, parent, NULL, name, namelen, 0,
                                     LUSTRE_OPC_ANY, NULL);
        if (IS_ERR(op_data))
                return PTR_ERR(op_data);

        op_data->op_valid = OBD_MD_FLID;
        rc = md_getattr_name(ll_i2sbi(parent)->ll_md_exp, op_data, &req);
        ll_finish_md_op_data(op_data);
        if (rc < 0)
                return rc;

        body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
        if (!body) {
                rc = -EFAULT;
                goto out_req;
        }
        if (fid)
                *fid = body->mbo_fid1;
out_req:
        ptlrpc_req_finished(req);
        return rc;
}

int ll_migrate(struct inode *parent, struct file *file, int mdtidx,
               const char *name, int namelen)
{
        struct ptlrpc_request *request = NULL;
        struct inode *child_inode = NULL;
        struct dentry *dchild = NULL;
        struct md_op_data *op_data;
        struct qstr qstr;
        int rc;

        CDEBUG(D_VFSTRACE, "migrate %s under "DFID" to MDT%d\n",
               name, PFID(ll_inode2fid(parent)), mdtidx);

        op_data = ll_prep_md_op_data(NULL, parent, NULL, name, namelen,
                                     0, LUSTRE_OPC_ANY, NULL);
        if (IS_ERR(op_data))
                return PTR_ERR(op_data);

        /* Get child FID first */
        qstr.hash = full_name_hash(parent, name, namelen);
        qstr.name = name;
        qstr.len = namelen;
        dchild = d_lookup(file_dentry(file), &qstr);
        if (dchild && dchild->d_inode) {
                op_data->op_fid3 = *ll_inode2fid(dchild->d_inode);
                if (dchild->d_inode) {
                        child_inode = igrab(dchild->d_inode);
                        ll_invalidate_aliases(child_inode);
                }
                dput(dchild);
        } else {
                rc = ll_get_fid_by_name(parent, name, namelen,
                                        &op_data->op_fid3);
                if (rc)
                        goto out_free;
        }

        if (!fid_is_sane(&op_data->op_fid3)) {
                CERROR("%s: migrate %s, but fid "DFID" is insane\n",
                       ll_get_fsname(parent->i_sb, NULL, 0), name,
                       PFID(&op_data->op_fid3));
                rc = -EINVAL;
                goto out_free;
        }

        rc = ll_get_mdt_idx_by_fid(ll_i2sbi(parent), &op_data->op_fid3);
        if (rc < 0)
                goto out_free;

        if (rc == mdtidx) {
                CDEBUG(D_INFO, "%s:"DFID" is already on MDT%d.\n", name,
                       PFID(&op_data->op_fid3), mdtidx);
                rc = 0;
                goto out_free;
        }

        op_data->op_mds = mdtidx;
        op_data->op_cli_flags = CLI_MIGRATE;
        rc = md_rename(ll_i2sbi(parent)->ll_md_exp, op_data, name,
                       namelen, name, namelen, &request);
        if (!rc)
                ll_update_times(request, parent);

        ptlrpc_req_finished(request);

out_free:
        if (child_inode) {
                clear_nlink(child_inode);
                iput(child_inode);
        }

        ll_finish_md_op_data(op_data);
        return rc;
}

static int
ll_file_noflock(struct file *file, int cmd, struct file_lock *file_lock)
{
        return -ENOSYS;
}

/**
 * test if some locks matching bits and l_req_mode are acquired
 * - bits can be in different locks
 * - if found clear the common lock bits in *bits
 * - the bits not found, are kept in *bits
 * \param inode [IN]
 * \param bits [IN] searched lock bits [IN]
 * \param l_req_mode [IN] searched lock mode
 * \retval boolean, true iff all bits are found
 */
int ll_have_md_lock(struct inode *inode, __u64 *bits,
                    enum ldlm_mode l_req_mode)
{
        struct lustre_handle lockh;
        ldlm_policy_data_t policy;
        enum ldlm_mode mode = (l_req_mode == LCK_MINMODE) ?
                              (LCK_CR | LCK_CW | LCK_PR | LCK_PW) : l_req_mode;
        struct lu_fid *fid;
        __u64 flags;
        int i;

        if (!inode)
                return 0;

        fid = &ll_i2info(inode)->lli_fid;
        CDEBUG(D_INFO, "trying to match res "DFID" mode %s\n", PFID(fid),
               ldlm_lockname[mode]);

        flags = LDLM_FL_BLOCK_GRANTED | LDLM_FL_CBPENDING | LDLM_FL_TEST_LOCK;
        for (i = 0; i <= MDS_INODELOCK_MAXSHIFT && *bits != 0; i++) {
                policy.l_inodebits.bits = *bits & (1 << i);
                if (policy.l_inodebits.bits == 0)
                        continue;

                if (md_lock_match(ll_i2mdexp(inode), flags, fid, LDLM_IBITS,
                                  &policy, mode, &lockh)) {
                        struct ldlm_lock *lock;

                        lock = ldlm_handle2lock(&lockh);
                        if (lock) {
                                *bits &=
                                      ~(lock->l_policy_data.l_inodebits.bits);
                                LDLM_LOCK_PUT(lock);
                        } else {
                                *bits &= ~policy.l_inodebits.bits;
                        }
                }
        }
        return *bits == 0;
}

enum ldlm_mode ll_take_md_lock(struct inode *inode, __u64 bits,
                               struct lustre_handle *lockh, __u64 flags,
                               enum ldlm_mode mode)
{
        ldlm_policy_data_t policy = { .l_inodebits = {bits} };
        struct lu_fid *fid;
        enum ldlm_mode rc;

        fid = &ll_i2info(inode)->lli_fid;
        CDEBUG(D_INFO, "trying to match res "DFID"\n", PFID(fid));

        rc = md_lock_match(ll_i2mdexp(inode), flags | LDLM_FL_BLOCK_GRANTED,
                           fid, LDLM_IBITS, &policy, mode, lockh);

        return rc;
}

static int ll_inode_revalidate_fini(struct inode *inode, int rc)
{
        /* Already unlinked. Just update nlink and return success */
        if (rc == -ENOENT) {
                clear_nlink(inode);
                /* This path cannot be hit for regular files unless in
                 * case of obscure races, so no need to validate size.
                 */
                if (!S_ISREG(inode->i_mode) && !S_ISDIR(inode->i_mode))
                        return 0;
        } else if (rc != 0) {
                CDEBUG_LIMIT((rc == -EACCES || rc == -EIDRM) ? D_INFO : D_ERROR,
                             "%s: revalidate FID "DFID" error: rc = %d\n",
                             ll_get_fsname(inode->i_sb, NULL, 0),
                             PFID(ll_inode2fid(inode)), rc);
        }

        return rc;
}

static int __ll_inode_revalidate(struct dentry *dentry, __u64 ibits)
{
        struct inode *inode = d_inode(dentry);
        struct ptlrpc_request *req = NULL;
        struct obd_export *exp;
        int rc = 0;

        CDEBUG(D_VFSTRACE, "VFS Op:inode="DFID"(%p),name=%pd\n",
               PFID(ll_inode2fid(inode)), inode, dentry);

        exp = ll_i2mdexp(inode);

        /* XXX: Enable OBD_CONNECT_ATTRFID to reduce unnecessary getattr RPC.
         *      But under CMD case, it caused some lock issues, should be fixed
         *      with new CMD ibits lock. See bug 12718
         */
        if (exp_connect_flags(exp) & OBD_CONNECT_ATTRFID) {
                struct lookup_intent oit = { .it_op = IT_GETATTR };
                struct md_op_data *op_data;

                if (ibits == MDS_INODELOCK_LOOKUP)
                        oit.it_op = IT_LOOKUP;

                /* Call getattr by fid, so do not provide name at all. */
                op_data = ll_prep_md_op_data(NULL, inode,
                                             inode, NULL, 0, 0,
                                             LUSTRE_OPC_ANY, NULL);
                if (IS_ERR(op_data))
                        return PTR_ERR(op_data);

                rc = md_intent_lock(exp, op_data, &oit, &req,
                                    &ll_md_blocking_ast, 0);
                ll_finish_md_op_data(op_data);
                if (rc < 0) {
                        rc = ll_inode_revalidate_fini(inode, rc);
                        goto out;
                }

                rc = ll_revalidate_it_finish(req, &oit, inode);
                if (rc != 0) {
                        ll_intent_release(&oit);
                        goto out;
                }

                /* Unlinked? Unhash dentry, so it is not picked up later by
                 * do_lookup() -> ll_revalidate_it(). We cannot use d_drop
                 * here to preserve get_cwd functionality on 2.6.
                 * Bug 10503
                 */
                if (!d_inode(dentry)->i_nlink) {
                        spin_lock(&inode->i_lock);
                        d_lustre_invalidate(dentry, 0);
                        spin_unlock(&inode->i_lock);
                }

                ll_lookup_finish_locks(&oit, inode);
        } else if (!ll_have_md_lock(d_inode(dentry), &ibits, LCK_MINMODE)) {
                struct ll_sb_info *sbi = ll_i2sbi(d_inode(dentry));
                u64 valid = OBD_MD_FLGETATTR;
                struct md_op_data *op_data;
                int ealen = 0;

                if (S_ISREG(inode->i_mode)) {
                        rc = ll_get_default_mdsize(sbi, &ealen);
                        if (rc)
                                return rc;
                        valid |= OBD_MD_FLEASIZE | OBD_MD_FLMODEASIZE;
                }

                op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL,
                                             0, ealen, LUSTRE_OPC_ANY,
                                             NULL);
                if (IS_ERR(op_data))
                        return PTR_ERR(op_data);

                op_data->op_valid = valid;
                rc = md_getattr(sbi->ll_md_exp, op_data, &req);
                ll_finish_md_op_data(op_data);
                if (rc) {
                        rc = ll_inode_revalidate_fini(inode, rc);
                        return rc;
                }

                rc = ll_prep_inode(&inode, req, NULL, NULL);
        }
out:
        ptlrpc_req_finished(req);
        return rc;
}

static int ll_merge_md_attr(struct inode *inode)
{
        struct cl_attr attr = { 0 };
        int rc;

        LASSERT(ll_i2info(inode)->lli_lsm_md);
        rc = md_merge_attr(ll_i2mdexp(inode), ll_i2info(inode)->lli_lsm_md,
                           &attr);
        if (rc)
                return rc;

        ll_i2info(inode)->lli_stripe_dir_size = attr.cat_size;
        ll_i2info(inode)->lli_stripe_dir_nlink = attr.cat_nlink;

        ll_i2info(inode)->lli_atime = attr.cat_atime;
        ll_i2info(inode)->lli_mtime = attr.cat_mtime;
        ll_i2info(inode)->lli_ctime = attr.cat_ctime;

        return 0;
}

static int ll_inode_revalidate(struct dentry *dentry, __u64 ibits)
{
        struct inode *inode = d_inode(dentry);
        int rc;

        rc = __ll_inode_revalidate(dentry, ibits);
        if (rc != 0)
                return rc;

        /* if object isn't regular file, don't validate size */
        if (!S_ISREG(inode->i_mode)) {
                if (S_ISDIR(inode->i_mode) &&
                    ll_i2info(inode)->lli_lsm_md) {
                        rc = ll_merge_md_attr(inode);
                        if (rc)
                                return rc;
                }

                LTIME_S(inode->i_atime) = ll_i2info(inode)->lli_atime;
                LTIME_S(inode->i_mtime) = ll_i2info(inode)->lli_mtime;
                LTIME_S(inode->i_ctime) = ll_i2info(inode)->lli_ctime;
        } else {
                /* In case of restore, the MDT has the right size and has
                 * already send it back without granting the layout lock,
                 * inode is up-to-date so glimpse is useless.
                 * Also to glimpse we need the layout, in case of a running
                 * restore the MDT holds the layout lock so the glimpse will
                 * block up to the end of restore (getattr will block)
                 */
                if (!(ll_i2info(inode)->lli_flags & LLIF_FILE_RESTORING))
                        rc = ll_glimpse_size(inode);
        }
        return rc;
}

int ll_getattr(struct vfsmount *mnt, struct dentry *de, struct kstat *stat)
{
        struct inode *inode = d_inode(de);
        struct ll_sb_info *sbi = ll_i2sbi(inode);
        struct ll_inode_info *lli = ll_i2info(inode);
        int res;

        res = ll_inode_revalidate(de, MDS_INODELOCK_UPDATE |
                                      MDS_INODELOCK_LOOKUP);
        ll_stats_ops_tally(sbi, LPROC_LL_GETATTR, 1);

        if (res)
                return res;

        OBD_FAIL_TIMEOUT(OBD_FAIL_GETATTR_DELAY, 30);

        stat->dev = inode->i_sb->s_dev;
        if (ll_need_32bit_api(sbi))
                stat->ino = cl_fid_build_ino(&lli->lli_fid, 1);
        else
                stat->ino = inode->i_ino;
        stat->mode = inode->i_mode;
        stat->uid = inode->i_uid;
        stat->gid = inode->i_gid;
        stat->rdev = inode->i_rdev;
        stat->atime = inode->i_atime;
        stat->mtime = inode->i_mtime;
        stat->ctime = inode->i_ctime;
        stat->blksize = 1 << inode->i_blkbits;
        stat->blocks = inode->i_blocks;

        if (S_ISDIR(inode->i_mode) &&
            ll_i2info(inode)->lli_lsm_md) {
                stat->nlink = lli->lli_stripe_dir_nlink;
                stat->size = lli->lli_stripe_dir_size;
        } else {
                stat->nlink = inode->i_nlink;
                stat->size = i_size_read(inode);
        }

        return 0;
}

static int ll_fiemap(struct inode *inode, struct fiemap_extent_info *fieinfo,
                     __u64 start, __u64 len)
{
        int rc;
        size_t num_bytes;
        struct ll_user_fiemap *fiemap;
        unsigned int extent_count = fieinfo->fi_extents_max;

        num_bytes = sizeof(*fiemap) + (extent_count *
                                       sizeof(struct ll_fiemap_extent));
        fiemap = libcfs_kvzalloc(num_bytes, GFP_NOFS);

        if (!fiemap)
                return -ENOMEM;

        fiemap->fm_flags = fieinfo->fi_flags;
        fiemap->fm_extent_count = fieinfo->fi_extents_max;
        fiemap->fm_start = start;
        fiemap->fm_length = len;
        if (extent_count > 0 &&
            copy_from_user(&fiemap->fm_extents[0], fieinfo->fi_extents_start,
                           sizeof(struct ll_fiemap_extent)) != 0) {
                rc = -EFAULT;
                goto out;
        }

        rc = ll_do_fiemap(inode, fiemap, num_bytes);

        fieinfo->fi_flags = fiemap->fm_flags;
        fieinfo->fi_extents_mapped = fiemap->fm_mapped_extents;
        if (extent_count > 0 &&
            copy_to_user(fieinfo->fi_extents_start, &fiemap->fm_extents[0],
                         fiemap->fm_mapped_extents *
                         sizeof(struct ll_fiemap_extent)) != 0) {
                rc = -EFAULT;
                goto out;
        }

out:
        kvfree(fiemap);
        return rc;
}

struct posix_acl *ll_get_acl(struct inode *inode, int type)
{
        struct ll_inode_info *lli = ll_i2info(inode);
        struct posix_acl *acl = NULL;

        spin_lock(&lli->lli_lock);
        /* VFS' acl_permission_check->check_acl will release the refcount */
        acl = posix_acl_dup(lli->lli_posix_acl);
#ifdef CONFIG_FS_POSIX_ACL
        forget_cached_acl(inode, type);
#endif
        spin_unlock(&lli->lli_lock);

        return acl;
}

int ll_inode_permission(struct inode *inode, int mask)
{
        struct ll_sb_info *sbi;
        struct root_squash_info *squash;
        const struct cred *old_cred = NULL;
        struct cred *cred = NULL;
        bool squash_id = false;
        cfs_cap_t cap;
        int rc = 0;

        if (mask & MAY_NOT_BLOCK)
                return -ECHILD;

       /* as root inode are NOT getting validated in lookup operation,
        * need to do it before permission check.
        */

        if (is_root_inode(inode)) {
                rc = __ll_inode_revalidate(inode->i_sb->s_root,
                                           MDS_INODELOCK_LOOKUP);
                if (rc)
                        return rc;
        }

        CDEBUG(D_VFSTRACE, "VFS Op:inode="DFID"(%p), inode mode %x mask %o\n",
               PFID(ll_inode2fid(inode)), inode, inode->i_mode, mask);

        /* squash fsuid/fsgid if needed */
        sbi = ll_i2sbi(inode);
        squash = &sbi->ll_squash;
        if (unlikely(squash->rsi_uid &&
                     uid_eq(current_fsuid(), GLOBAL_ROOT_UID) &&
                     !(sbi->ll_flags & LL_SBI_NOROOTSQUASH))) {
                squash_id = true;
        }

        if (squash_id) {
                CDEBUG(D_OTHER, "squash creds (%d:%d)=>(%d:%d)\n",
                       __kuid_val(current_fsuid()), __kgid_val(current_fsgid()),
                       squash->rsi_uid, squash->rsi_gid);

                /*
                 * update current process's credentials
                 * and FS capability
                 */
                cred = prepare_creds();
                if (!cred)
                        return -ENOMEM;

                cred->fsuid = make_kuid(&init_user_ns, squash->rsi_uid);
                cred->fsgid = make_kgid(&init_user_ns, squash->rsi_gid);
                for (cap = 0; cap < sizeof(cfs_cap_t) * 8; cap++) {
                        if ((1 << cap) & CFS_CAP_FS_MASK)
                                cap_lower(cred->cap_effective, cap);
                }
                old_cred = override_creds(cred);
        }

        ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_INODE_PERM, 1);
        rc = generic_permission(inode, mask);

        /* restore current process's credentials and FS capability */
        if (squash_id) {
                revert_creds(old_cred);
                put_cred(cred);
        }

        return rc;
}

/* -o localflock - only provides locally consistent flock locks */
struct file_operations ll_file_operations = {
        .read_iter = ll_file_read_iter,
        .write_iter = ll_file_write_iter,
        .unlocked_ioctl = ll_file_ioctl,
        .open      = ll_file_open,
        .release        = ll_file_release,
        .mmap      = ll_file_mmap,
        .llseek  = ll_file_seek,
        .splice_read    = ll_file_splice_read,
        .fsync    = ll_fsync,
        .flush    = ll_flush
};

struct file_operations ll_file_operations_flock = {
        .read_iter    = ll_file_read_iter,
        .write_iter   = ll_file_write_iter,
        .unlocked_ioctl = ll_file_ioctl,
        .open      = ll_file_open,
        .release        = ll_file_release,
        .mmap      = ll_file_mmap,
        .llseek  = ll_file_seek,
        .splice_read    = ll_file_splice_read,
        .fsync    = ll_fsync,
        .flush    = ll_flush,
        .flock    = ll_file_flock,
        .lock      = ll_file_flock
};

/* These are for -o noflock - to return ENOSYS on flock calls */
struct file_operations ll_file_operations_noflock = {
        .read_iter    = ll_file_read_iter,
        .write_iter   = ll_file_write_iter,
        .unlocked_ioctl = ll_file_ioctl,
        .open      = ll_file_open,
        .release        = ll_file_release,
        .mmap      = ll_file_mmap,
        .llseek  = ll_file_seek,
        .splice_read    = ll_file_splice_read,
        .fsync    = ll_fsync,
        .flush    = ll_flush,
        .flock    = ll_file_noflock,
        .lock      = ll_file_noflock
};

const struct inode_operations ll_file_inode_operations = {
        .setattr        = ll_setattr,
        .getattr        = ll_getattr,
        .permission     = ll_inode_permission,
        .setxattr       = generic_setxattr,
        .getxattr       = generic_getxattr,
        .listxattr      = ll_listxattr,
        .removexattr    = generic_removexattr,
        .fiemap         = ll_fiemap,
        .get_acl        = ll_get_acl,
};

/* dynamic ioctl number support routines */
static struct llioc_ctl_data {
        struct rw_semaphore     ioc_sem;
        struct list_head              ioc_head;
} llioc = {
        __RWSEM_INITIALIZER(llioc.ioc_sem),
        LIST_HEAD_INIT(llioc.ioc_head)
};

struct llioc_data {
        struct list_head              iocd_list;
        unsigned int        iocd_size;
        llioc_callback_t        iocd_cb;
        unsigned int        iocd_count;
        unsigned int        iocd_cmd[0];
};

void *ll_iocontrol_register(llioc_callback_t cb, int count, unsigned int *cmd)
{
        unsigned int size;
        struct llioc_data *in_data = NULL;

        if (!cb || !cmd || count > LLIOC_MAX_CMD || count < 0)
                return NULL;

        size = sizeof(*in_data) + count * sizeof(unsigned int);
        in_data = kzalloc(size, GFP_NOFS);
        if (!in_data)
                return NULL;

        in_data->iocd_size = size;
        in_data->iocd_cb = cb;
        in_data->iocd_count = count;
        memcpy(in_data->iocd_cmd, cmd, sizeof(unsigned int) * count);

        down_write(&llioc.ioc_sem);
        list_add_tail(&in_data->iocd_list, &llioc.ioc_head);
        up_write(&llioc.ioc_sem);

        return in_data;
}
EXPORT_SYMBOL(ll_iocontrol_register);

void ll_iocontrol_unregister(void *magic)
{
        struct llioc_data *tmp;

        if (!magic)
                return;

        down_write(&llioc.ioc_sem);
        list_for_each_entry(tmp, &llioc.ioc_head, iocd_list) {
                if (tmp == magic) {
                        list_del(&tmp->iocd_list);
                        up_write(&llioc.ioc_sem);

                        kfree(tmp);
                        return;
                }
        }
        up_write(&llioc.ioc_sem);

        CWARN("didn't find iocontrol register block with magic: %p\n", magic);
}
EXPORT_SYMBOL(ll_iocontrol_unregister);

static enum llioc_iter
ll_iocontrol_call(struct inode *inode, struct file *file,
                  unsigned int cmd, unsigned long arg, int *rcp)
{
        enum llioc_iter ret = LLIOC_CONT;
        struct llioc_data *data;
        int rc = -EINVAL, i;

        down_read(&llioc.ioc_sem);
        list_for_each_entry(data, &llioc.ioc_head, iocd_list) {
                for (i = 0; i < data->iocd_count; i++) {
                        if (cmd != data->iocd_cmd[i])
                                continue;

                        ret = data->iocd_cb(inode, file, cmd, arg, data, &rc);
                        break;
                }

                if (ret == LLIOC_STOP)
                        break;
        }
        up_read(&llioc.ioc_sem);

        if (rcp)
                *rcp = rc;
        return ret;
}

int ll_layout_conf(struct inode *inode, const struct cl_object_conf *conf)
{
        struct ll_inode_info *lli = ll_i2info(inode);
        struct cl_env_nest nest;
        struct lu_env *env;
        int result;

        if (!lli->lli_clob)
                return 0;

        env = cl_env_nested_get(&nest);
        if (IS_ERR(env))
                return PTR_ERR(env);

        result = cl_conf_set(env, lli->lli_clob, conf);
        cl_env_nested_put(&nest, env);

        if (conf->coc_opc == OBJECT_CONF_SET) {
                struct ldlm_lock *lock = conf->coc_lock;

                LASSERT(lock);
                LASSERT(ldlm_has_layout(lock));
                if (result == 0) {
                        /* it can only be allowed to match after layout is
                         * applied to inode otherwise false layout would be
                         * seen. Applying layout should happen before dropping
                         * the intent lock.
                         */
                        ldlm_lock_allow_match(lock);
                }
        }
        return result;
}

/* Fetch layout from MDT with getxattr request, if it's not ready yet */
static int ll_layout_fetch(struct inode *inode, struct ldlm_lock *lock)

{
        struct ll_sb_info *sbi = ll_i2sbi(inode);
        struct ptlrpc_request *req;
        struct mdt_body *body;
        void *lvbdata;
        void *lmm;
        int lmmsize;
        int rc;

        CDEBUG(D_INODE, DFID" LVB_READY=%d l_lvb_data=%p l_lvb_len=%d\n",
               PFID(ll_inode2fid(inode)), ldlm_is_lvb_ready(lock),
               lock->l_lvb_data, lock->l_lvb_len);

        if (lock->l_lvb_data && ldlm_is_lvb_ready(lock))
                return 0;

        /* if layout lock was granted right away, the layout is returned
         * within DLM_LVB of dlm reply; otherwise if the lock was ever
         * blocked and then granted via completion ast, we have to fetch
         * layout here. Please note that we can't use the LVB buffer in
         * completion AST because it doesn't have a large enough buffer
         */
        rc = ll_get_default_mdsize(sbi, &lmmsize);
        if (rc == 0)
                rc = md_getxattr(sbi->ll_md_exp, ll_inode2fid(inode),
                                 OBD_MD_FLXATTR, XATTR_NAME_LOV, NULL, 0,
                                 lmmsize, 0, &req);
        if (rc < 0)
                return rc;

        body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
        if (!body) {
                rc = -EPROTO;
                goto out;
        }

        lmmsize = body->mbo_eadatasize;
        if (lmmsize == 0) /* empty layout */ {
                rc = 0;
                goto out;
        }

        lmm = req_capsule_server_sized_get(&req->rq_pill, &RMF_EADATA, lmmsize);
        if (!lmm) {
                rc = -EFAULT;
                goto out;
        }

        lvbdata = libcfs_kvzalloc(lmmsize, GFP_NOFS);
        if (!lvbdata) {
                rc = -ENOMEM;
                goto out;
        }

        memcpy(lvbdata, lmm, lmmsize);
        lock_res_and_lock(lock);
        if (lock->l_lvb_data)
                kvfree(lock->l_lvb_data);

        lock->l_lvb_data = lvbdata;
        lock->l_lvb_len = lmmsize;
        unlock_res_and_lock(lock);

out:
        ptlrpc_req_finished(req);
        return rc;
}

/**
 * Apply the layout to the inode. Layout lock is held and will be released
 * in this function.
 */
static int ll_layout_lock_set(struct lustre_handle *lockh, enum ldlm_mode mode,
                              struct inode *inode, __u32 *gen, bool reconf)
{
        struct ll_inode_info *lli = ll_i2info(inode);
        struct ll_sb_info    *sbi = ll_i2sbi(inode);
        struct ldlm_lock *lock;
        struct lustre_md md = { NULL };
        struct cl_object_conf conf;
        int rc = 0;
        bool lvb_ready;
        bool wait_layout = false;

        LASSERT(lustre_handle_is_used(lockh));

        lock = ldlm_handle2lock(lockh);
        LASSERT(lock);
        LASSERT(ldlm_has_layout(lock));

        LDLM_DEBUG(lock, "File "DFID"(%p) being reconfigured: %d",
                   PFID(&lli->lli_fid), inode, reconf);

        /* in case this is a caching lock and reinstate with new inode */
        md_set_lock_data(sbi->ll_md_exp, lockh, inode, NULL);

        lock_res_and_lock(lock);
        lvb_ready = ldlm_is_lvb_ready(lock);
        unlock_res_and_lock(lock);
        /* checking lvb_ready is racy but this is okay. The worst case is
         * that multi processes may configure the file on the same time.
         */
        if (lvb_ready || !reconf) {
                rc = -ENODATA;
                if (lvb_ready) {
                        /* layout_gen must be valid if layout lock is not
                         * cancelled and stripe has already set
                         */
                        *gen = ll_layout_version_get(lli);
                        rc = 0;
                }
                goto out;
        }

        rc = ll_layout_fetch(inode, lock);
        if (rc < 0)
                goto out;

        /* for layout lock, lmm is returned in lock's lvb.
         * lvb_data is immutable if the lock is held so it's safe to access it
         * without res lock. See the description in ldlm_lock_decref_internal()
         * for the condition to free lvb_data of layout lock
         */
        if (lock->l_lvb_data) {
                rc = obd_unpackmd(sbi->ll_dt_exp, &md.lsm,
                                  lock->l_lvb_data, lock->l_lvb_len);
                if (rc >= 0) {
                        *gen = LL_LAYOUT_GEN_EMPTY;
                        if (md.lsm)
                                *gen = md.lsm->lsm_layout_gen;
                        rc = 0;
                } else {
                        CERROR("%s: file " DFID " unpackmd error: %d\n",
                               ll_get_fsname(inode->i_sb, NULL, 0),
                               PFID(&lli->lli_fid), rc);
                }
        }
        if (rc < 0)
                goto out;

        /* set layout to file. Unlikely this will fail as old layout was
         * surely eliminated
         */
        memset(&conf, 0, sizeof(conf));
        conf.coc_opc = OBJECT_CONF_SET;
        conf.coc_inode = inode;
        conf.coc_lock = lock;
        conf.u.coc_md = &md;
        rc = ll_layout_conf(inode, &conf);

        if (md.lsm)
                obd_free_memmd(sbi->ll_dt_exp, &md.lsm);

        /* refresh layout failed, need to wait */
        wait_layout = rc == -EBUSY;

out:
        LDLM_LOCK_PUT(lock);
        ldlm_lock_decref(lockh, mode);

        /* wait for IO to complete if it's still being used. */
        if (wait_layout) {
                CDEBUG(D_INODE, "%s: "DFID"(%p) wait for layout reconf\n",
                       ll_get_fsname(inode->i_sb, NULL, 0),
                       PFID(&lli->lli_fid), inode);

                memset(&conf, 0, sizeof(conf));
                conf.coc_opc = OBJECT_CONF_WAIT;
                conf.coc_inode = inode;
                rc = ll_layout_conf(inode, &conf);
                if (rc == 0)
                        rc = -EAGAIN;

                CDEBUG(D_INODE, "%s: file="DFID" waiting layout return: %d.\n",
                       ll_get_fsname(inode->i_sb, NULL, 0),
                       PFID(&lli->lli_fid), rc);
        }
        return rc;
}

/**
 * This function checks if there exists a LAYOUT lock on the client side,
 * or enqueues it if it doesn't have one in cache.
 *
 * This function will not hold layout lock so it may be revoked any time after
 * this function returns. Any operations depend on layout should be redone
 * in that case.
 *
 * This function should be called before lov_io_init() to get an uptodate
 * layout version, the caller should save the version number and after IO
 * is finished, this function should be called again to verify that layout
 * is not changed during IO time.
 */
int ll_layout_refresh(struct inode *inode, __u32 *gen)
{
        struct ll_inode_info  *lli = ll_i2info(inode);
        struct ll_sb_info     *sbi = ll_i2sbi(inode);
        struct md_op_data     *op_data;
        struct lookup_intent   it;
        struct lustre_handle   lockh;
        enum ldlm_mode         mode;
        struct ldlm_enqueue_info einfo = {
                .ei_type = LDLM_IBITS,
                .ei_mode = LCK_CR,
                .ei_cb_bl = &ll_md_blocking_ast,
                .ei_cb_cp = &ldlm_completion_ast,
        };
        int rc;

        *gen = ll_layout_version_get(lli);
        if (!(sbi->ll_flags & LL_SBI_LAYOUT_LOCK) || *gen != LL_LAYOUT_GEN_NONE)
                return 0;

        /* sanity checks */
        LASSERT(fid_is_sane(ll_inode2fid(inode)));
        LASSERT(S_ISREG(inode->i_mode));

        /* take layout lock mutex to enqueue layout lock exclusively. */
        mutex_lock(&lli->lli_layout_mutex);

again:
        /* mostly layout lock is caching on the local side, so try to match
         * it before grabbing layout lock mutex.
         */
        mode = ll_take_md_lock(inode, MDS_INODELOCK_LAYOUT, &lockh, 0,
                               LCK_CR | LCK_CW | LCK_PR | LCK_PW);
        if (mode != 0) { /* hit cached lock */
                rc = ll_layout_lock_set(&lockh, mode, inode, gen, true);
                if (rc == -EAGAIN)
                        goto again;

                mutex_unlock(&lli->lli_layout_mutex);
                return rc;
        }

        op_data = ll_prep_md_op_data(NULL, inode, inode, NULL,
                                     0, 0, LUSTRE_OPC_ANY, NULL);
        if (IS_ERR(op_data)) {
                mutex_unlock(&lli->lli_layout_mutex);
                return PTR_ERR(op_data);
        }

        /* have to enqueue one */
        memset(&it, 0, sizeof(it));
        it.it_op = IT_LAYOUT;
        lockh.cookie = 0ULL;

        LDLM_DEBUG_NOLOCK("%s: requeue layout lock for file "DFID"(%p)",
                          ll_get_fsname(inode->i_sb, NULL, 0),
                          PFID(&lli->lli_fid), inode);

        rc = md_enqueue(sbi->ll_md_exp, &einfo, NULL, &it, op_data, &lockh, 0);
        ptlrpc_req_finished(it.it_request);
        it.it_request = NULL;

        ll_finish_md_op_data(op_data);

        mode = it.it_lock_mode;
        it.it_lock_mode = 0;
        ll_intent_drop_lock(&it);

        if (rc == 0) {
                /* set lock data in case this is a new lock */
                ll_set_lock_data(sbi->ll_md_exp, inode, &it, NULL);
                rc = ll_layout_lock_set(&lockh, mode, inode, gen, true);
                if (rc == -EAGAIN)
                        goto again;
        }
        mutex_unlock(&lli->lli_layout_mutex);

        return rc;
}

/**
 *  This function send a restore request to the MDT
 */
int ll_layout_restore(struct inode *inode, loff_t offset, __u64 length)
{
        struct hsm_user_request *hur;
        int                      len, rc;

        len = sizeof(struct hsm_user_request) +
              sizeof(struct hsm_user_item);
        hur = kzalloc(len, GFP_NOFS);
        if (!hur)
                return -ENOMEM;

        hur->hur_request.hr_action = HUA_RESTORE;
        hur->hur_request.hr_archive_id = 0;
        hur->hur_request.hr_flags = 0;
        memcpy(&hur->hur_user_item[0].hui_fid, &ll_i2info(inode)->lli_fid,
               sizeof(hur->hur_user_item[0].hui_fid));
        hur->hur_user_item[0].hui_extent.offset = offset;
        hur->hur_user_item[0].hui_extent.length = length;
        hur->hur_request.hr_itemcount = 1;
        rc = obd_iocontrol(LL_IOC_HSM_REQUEST, ll_i2sbi(inode)->ll_md_exp,
                           len, hur, NULL);
        kfree(hur);
        return rc;
}