/*
 * 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) 2003, 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.
 */

#define DEBUG_SUBSYSTEM S_MDC

# include <linux/module.h>

#include "../include/lustre_intent.h"
#include "../include/obd.h"
#include "../include/obd_class.h"
#include "../include/lustre_dlm.h"
#include "../include/lustre_fid.h"
#include "../include/lustre_mdc.h"
#include "../include/lustre_net.h"
#include "../include/lustre_req_layout.h"
#include "../include/lustre_swab.h"

#include "mdc_internal.h"

struct mdc_getattr_args {
        struct obd_export          *ga_exp;
        struct md_enqueue_info      *ga_minfo;
};

int it_open_error(int phase, struct lookup_intent *it)
{
        if (it_disposition(it, DISP_OPEN_LEASE)) {
                if (phase >= DISP_OPEN_LEASE)
                        return it->it_status;
                else
                        return 0;
        }
        if (it_disposition(it, DISP_OPEN_OPEN)) {
                if (phase >= DISP_OPEN_OPEN)
                        return it->it_status;
                else
                        return 0;
        }

        if (it_disposition(it, DISP_OPEN_CREATE)) {
                if (phase >= DISP_OPEN_CREATE)
                        return it->it_status;
                else
                        return 0;
        }

        if (it_disposition(it, DISP_LOOKUP_EXECD)) {
                if (phase >= DISP_LOOKUP_EXECD)
                        return it->it_status;
                else
                        return 0;
        }

        if (it_disposition(it, DISP_IT_EXECD)) {
                if (phase >= DISP_IT_EXECD)
                        return it->it_status;
                else
                        return 0;
        }
        CERROR("it disp: %X, status: %d\n", it->it_disposition,
               it->it_status);
        LBUG();
        return 0;
}
EXPORT_SYMBOL(it_open_error);

/* this must be called on a lockh that is known to have a referenced lock */
int mdc_set_lock_data(struct obd_export *exp, const struct lustre_handle *lockh,
                      void *data, __u64 *bits)
{
        struct ldlm_lock *lock;
        struct inode *new_inode = data;

        if (bits)
                *bits = 0;

        if (!lustre_handle_is_used(lockh))
                return 0;

        lock = ldlm_handle2lock(lockh);

        LASSERT(lock);
        lock_res_and_lock(lock);
        if (lock->l_resource->lr_lvb_inode &&
            lock->l_resource->lr_lvb_inode != data) {
                struct inode *old_inode = lock->l_resource->lr_lvb_inode;

                LASSERTF(old_inode->i_state & I_FREEING,
                         "Found existing inode %p/%lu/%u state %lu in lock: setting data to %p/%lu/%u\n",
                         old_inode, old_inode->i_ino, old_inode->i_generation,
                         old_inode->i_state, new_inode, new_inode->i_ino,
                         new_inode->i_generation);
        }
        lock->l_resource->lr_lvb_inode = new_inode;
        if (bits)
                *bits = lock->l_policy_data.l_inodebits.bits;

        unlock_res_and_lock(lock);
        LDLM_LOCK_PUT(lock);

        return 0;
}

enum ldlm_mode mdc_lock_match(struct obd_export *exp, __u64 flags,
                              const struct lu_fid *fid, enum ldlm_type type,
                              union ldlm_policy_data *policy,
                              enum ldlm_mode mode,
                              struct lustre_handle *lockh)
{
        struct ldlm_res_id res_id;
        enum ldlm_mode rc;

        fid_build_reg_res_name(fid, &res_id);
        /* LU-4405: Clear bits not supported by server */
        policy->l_inodebits.bits &= exp_connect_ibits(exp);
        rc = ldlm_lock_match(class_exp2obd(exp)->obd_namespace, flags,
                             &res_id, type, policy, mode, lockh, 0);
        return rc;
}

int mdc_cancel_unused(struct obd_export *exp,
                      const struct lu_fid *fid,
                      union ldlm_policy_data *policy,
                      enum ldlm_mode mode,
                      enum ldlm_cancel_flags flags,
                      void *opaque)
{
        struct ldlm_res_id res_id;
        struct obd_device *obd = class_exp2obd(exp);
        int rc;

        fid_build_reg_res_name(fid, &res_id);
        rc = ldlm_cli_cancel_unused_resource(obd->obd_namespace, &res_id,
                                             policy, mode, flags, opaque);
        return rc;
}

int mdc_null_inode(struct obd_export *exp,
                   const struct lu_fid *fid)
{
        struct ldlm_res_id res_id;
        struct ldlm_resource *res;
        struct ldlm_namespace *ns = class_exp2obd(exp)->obd_namespace;

        LASSERTF(ns, "no namespace passed\n");

        fid_build_reg_res_name(fid, &res_id);

        res = ldlm_resource_get(ns, NULL, &res_id, 0, 0);
        if (IS_ERR(res))
                return 0;

        lock_res(res);
        res->lr_lvb_inode = NULL;
        unlock_res(res);

        ldlm_resource_putref(res);
        return 0;
}

static inline void mdc_clear_replay_flag(struct ptlrpc_request *req, int rc)
{
        /* Don't hold error requests for replay. */
        if (req->rq_replay) {
                spin_lock(&req->rq_lock);
                req->rq_replay = 0;
                spin_unlock(&req->rq_lock);
        }
        if (rc && req->rq_transno != 0) {
                DEBUG_REQ(D_ERROR, req, "transno returned on error rc %d", rc);
                LBUG();
        }
}

/* Save a large LOV EA into the request buffer so that it is available
 * for replay.  We don't do this in the initial request because the
 * original request doesn't need this buffer (at most it sends just the
 * lov_mds_md) and it is a waste of RAM/bandwidth to send the empty
 * buffer and may also be difficult to allocate and save a very large
 * request buffer for each open. (bug 5707)
 *
 * OOM here may cause recovery failure if lmm is needed (only for the
 * original open if the MDS crashed just when this client also OOM'd)
 * but this is incredibly unlikely, and questionable whether the client
 * could do MDS recovery under OOM anyways...
 */
static void mdc_realloc_openmsg(struct ptlrpc_request *req,
                                struct mdt_body *body)
{
        int     rc;

        /* FIXME: remove this explicit offset. */
        rc = sptlrpc_cli_enlarge_reqbuf(req, DLM_INTENT_REC_OFF + 4,
                                        body->mbo_eadatasize);
        if (rc) {
                CERROR("Can't enlarge segment %d size to %d\n",
                       DLM_INTENT_REC_OFF + 4, body->mbo_eadatasize);
                body->mbo_valid &= ~OBD_MD_FLEASIZE;
                body->mbo_eadatasize = 0;
        }
}

static struct ptlrpc_request *
mdc_intent_open_pack(struct obd_export *exp, struct lookup_intent *it,
                     struct md_op_data *op_data)
{
        struct ptlrpc_request *req;
        struct obd_device     *obddev = class_exp2obd(exp);
        struct ldlm_intent    *lit;
        const void *lmm = op_data->op_data;
        u32 lmmsize = op_data->op_data_size;
        LIST_HEAD(cancels);
        int                 count = 0;
        int                 mode;
        int                 rc;

        it->it_create_mode = (it->it_create_mode & ~S_IFMT) | S_IFREG;

        /* XXX: openlock is not cancelled for cross-refs. */
        /* If inode is known, cancel conflicting OPEN locks. */
        if (fid_is_sane(&op_data->op_fid2)) {
                if (it->it_flags & MDS_OPEN_LEASE) { /* try to get lease */
                        if (it->it_flags & FMODE_WRITE)
                                mode = LCK_EX;
                        else
                                mode = LCK_PR;
                } else {
                        if (it->it_flags & (FMODE_WRITE | MDS_OPEN_TRUNC))
                                mode = LCK_CW;
                        else if (it->it_flags & __FMODE_EXEC)
                                mode = LCK_PR;
                        else
                                mode = LCK_CR;
                }
                count = mdc_resource_get_unused(exp, &op_data->op_fid2,
                                                &cancels, mode,
                                                MDS_INODELOCK_OPEN);
        }

        /* If CREATE, cancel parent's UPDATE lock. */
        if (it->it_op & IT_CREAT)
                mode = LCK_EX;
        else
                mode = LCK_CR;
        count += mdc_resource_get_unused(exp, &op_data->op_fid1,
                                         &cancels, mode,
                                         MDS_INODELOCK_UPDATE);

        req = ptlrpc_request_alloc(class_exp2cliimp(exp),
                                   &RQF_LDLM_INTENT_OPEN);
        if (!req) {
                ldlm_lock_list_put(&cancels, l_bl_ast, count);
                return ERR_PTR(-ENOMEM);
        }

        req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
                             op_data->op_namelen + 1);
        req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT,
                             max(lmmsize, obddev->u.cli.cl_default_mds_easize));

        rc = ldlm_prep_enqueue_req(exp, req, &cancels, count);
        if (rc < 0) {
                ptlrpc_request_free(req);
                return ERR_PTR(rc);
        }

        spin_lock(&req->rq_lock);
        req->rq_replay = req->rq_import->imp_replayable;
        spin_unlock(&req->rq_lock);

        /* pack the intent */
        lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
        lit->opc = (__u64)it->it_op;

        /* pack the intended request */
        mdc_open_pack(req, op_data, it->it_create_mode, 0, it->it_flags, lmm,
                      lmmsize);

        req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
                             obddev->u.cli.cl_max_mds_easize);

        ptlrpc_request_set_replen(req);
        return req;
}

static struct ptlrpc_request *
mdc_intent_getxattr_pack(struct obd_export *exp,
                         struct lookup_intent *it,
                         struct md_op_data *op_data)
{
        struct ptlrpc_request   *req;
        struct ldlm_intent      *lit;
        int rc, count = 0;
        u32 maxdata;
        LIST_HEAD(cancels);

        req = ptlrpc_request_alloc(class_exp2cliimp(exp),
                                   &RQF_LDLM_INTENT_GETXATTR);
        if (!req)
                return ERR_PTR(-ENOMEM);

        rc = ldlm_prep_enqueue_req(exp, req, &cancels, count);
        if (rc) {
                ptlrpc_request_free(req);
                return ERR_PTR(rc);
        }

        /* pack the intent */
        lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
        lit->opc = IT_GETXATTR;

        maxdata = class_exp2cliimp(exp)->imp_connect_data.ocd_max_easize;

        /* pack the intended request */
        mdc_pack_body(req, &op_data->op_fid1, op_data->op_valid, maxdata, -1,
                      0);

        req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_SERVER, maxdata);

        req_capsule_set_size(&req->rq_pill, &RMF_EAVALS, RCL_SERVER, maxdata);

        req_capsule_set_size(&req->rq_pill, &RMF_EAVALS_LENS,
                             RCL_SERVER, maxdata);

        ptlrpc_request_set_replen(req);

        return req;
}

static struct ptlrpc_request *mdc_intent_unlink_pack(struct obd_export *exp,
                                                     struct lookup_intent *it,
                                                     struct md_op_data *op_data)
{
        struct ptlrpc_request *req;
        struct obd_device     *obddev = class_exp2obd(exp);
        struct ldlm_intent    *lit;
        int                 rc;

        req = ptlrpc_request_alloc(class_exp2cliimp(exp),
                                   &RQF_LDLM_INTENT_UNLINK);
        if (!req)
                return ERR_PTR(-ENOMEM);

        req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
                             op_data->op_namelen + 1);

        rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
        if (rc) {
                ptlrpc_request_free(req);
                return ERR_PTR(rc);
        }

        /* pack the intent */
        lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
        lit->opc = (__u64)it->it_op;

        /* pack the intended request */
        mdc_unlink_pack(req, op_data);

        req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
                             obddev->u.cli.cl_default_mds_easize);
        ptlrpc_request_set_replen(req);
        return req;
}

static struct ptlrpc_request *mdc_intent_getattr_pack(struct obd_export *exp,
                                                      struct lookup_intent *it,
                                                     struct md_op_data *op_data)
{
        struct ptlrpc_request *req;
        struct obd_device     *obddev = class_exp2obd(exp);
        u64                    valid = OBD_MD_FLGETATTR | OBD_MD_FLEASIZE |
                                       OBD_MD_FLMODEASIZE | OBD_MD_FLDIREA |
                                       OBD_MD_MEA | OBD_MD_FLACL;
        struct ldlm_intent    *lit;
        int                 rc;
        u32 easize;

        req = ptlrpc_request_alloc(class_exp2cliimp(exp),
                                   &RQF_LDLM_INTENT_GETATTR);
        if (!req)
                return ERR_PTR(-ENOMEM);

        req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
                             op_data->op_namelen + 1);

        rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
        if (rc) {
                ptlrpc_request_free(req);
                return ERR_PTR(rc);
        }

        /* pack the intent */
        lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
        lit->opc = (__u64)it->it_op;

        if (obddev->u.cli.cl_default_mds_easize > 0)
                easize = obddev->u.cli.cl_default_mds_easize;
        else
                easize = obddev->u.cli.cl_max_mds_easize;

        /* pack the intended request */
        mdc_getattr_pack(req, valid, it->it_flags, op_data, easize);

        req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER, easize);
        ptlrpc_request_set_replen(req);
        return req;
}

static struct ptlrpc_request *mdc_intent_layout_pack(struct obd_export *exp,
                                                     struct lookup_intent *it,
                                                     struct md_op_data *unused)
{
        struct obd_device     *obd = class_exp2obd(exp);
        struct ptlrpc_request *req;
        struct ldlm_intent    *lit;
        struct layout_intent  *layout;
        int rc;

        req = ptlrpc_request_alloc(class_exp2cliimp(exp),
                                   &RQF_LDLM_INTENT_LAYOUT);
        if (!req)
                return ERR_PTR(-ENOMEM);

        req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT, 0);
        rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
        if (rc) {
                ptlrpc_request_free(req);
                return ERR_PTR(rc);
        }

        /* pack the intent */
        lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
        lit->opc = (__u64)it->it_op;

        /* pack the layout intent request */
        layout = req_capsule_client_get(&req->rq_pill, &RMF_LAYOUT_INTENT);
        /* LAYOUT_INTENT_ACCESS is generic, specific operation will be
         * set for replication
         */
        layout->li_opc = LAYOUT_INTENT_ACCESS;

        req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
                             obd->u.cli.cl_default_mds_easize);
        ptlrpc_request_set_replen(req);
        return req;
}

static struct ptlrpc_request *
mdc_enqueue_pack(struct obd_export *exp, int lvb_len)
{
        struct ptlrpc_request *req;
        int rc;

        req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LDLM_ENQUEUE);
        if (!req)
                return ERR_PTR(-ENOMEM);

        rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
        if (rc) {
                ptlrpc_request_free(req);
                return ERR_PTR(rc);
        }

        req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER, lvb_len);
        ptlrpc_request_set_replen(req);
        return req;
}

static int mdc_finish_enqueue(struct obd_export *exp,
                              struct ptlrpc_request *req,
                              struct ldlm_enqueue_info *einfo,
                              struct lookup_intent *it,
                              struct lustre_handle *lockh,
                              int rc)
{
        struct req_capsule  *pill = &req->rq_pill;
        struct ldlm_request *lockreq;
        struct ldlm_reply   *lockrep;
        struct ldlm_lock    *lock;
        void            *lvb_data = NULL;
        u32 lvb_len = 0;

        LASSERT(rc >= 0);
        /* Similarly, if we're going to replay this request, we don't want to
         * actually get a lock, just perform the intent.
         */
        if (req->rq_transno || req->rq_replay) {
                lockreq = req_capsule_client_get(pill, &RMF_DLM_REQ);
                lockreq->lock_flags |= ldlm_flags_to_wire(LDLM_FL_INTENT_ONLY);
        }

        if (rc == ELDLM_LOCK_ABORTED) {
                einfo->ei_mode = 0;
                memset(lockh, 0, sizeof(*lockh));
                rc = 0;
        } else { /* rc = 0 */
                lock = ldlm_handle2lock(lockh);

                /* If the server gave us back a different lock mode, we should
                 * fix up our variables.
                 */
                if (lock->l_req_mode != einfo->ei_mode) {
                        ldlm_lock_addref(lockh, lock->l_req_mode);
                        ldlm_lock_decref(lockh, einfo->ei_mode);
                        einfo->ei_mode = lock->l_req_mode;
                }
                LDLM_LOCK_PUT(lock);
        }

        lockrep = req_capsule_server_get(pill, &RMF_DLM_REP);

        it->it_disposition = (int)lockrep->lock_policy_res1;
        it->it_status = (int)lockrep->lock_policy_res2;
        it->it_lock_mode = einfo->ei_mode;
        it->it_lock_handle = lockh->cookie;
        it->it_request = req;

        /* Technically speaking rq_transno must already be zero if
         * it_status is in error, so the check is a bit redundant
         */
        if ((!req->rq_transno || it->it_status < 0) && req->rq_replay)
                mdc_clear_replay_flag(req, it->it_status);

        /* If we're doing an IT_OPEN which did not result in an actual
         * successful open, then we need to remove the bit which saves
         * this request for unconditional replay.
         *
         * It's important that we do this first!  Otherwise we might exit the
         * function without doing so, and try to replay a failed create
         * (bug 3440)
         */
        if (it->it_op & IT_OPEN && req->rq_replay &&
            (!it_disposition(it, DISP_OPEN_OPEN) || it->it_status != 0))
                mdc_clear_replay_flag(req, it->it_status);

        DEBUG_REQ(D_RPCTRACE, req, "op: %d disposition: %x, status: %d",
                  it->it_op, it->it_disposition, it->it_status);

        /* We know what to expect, so we do any byte flipping required here */
        if (it->it_op & (IT_OPEN | IT_UNLINK | IT_LOOKUP | IT_GETATTR)) {
                struct mdt_body *body;

                body = req_capsule_server_get(pill, &RMF_MDT_BODY);
                if (!body) {
                        CERROR("Can't swab mdt_body\n");
                        return -EPROTO;
                }

                if (it_disposition(it, DISP_OPEN_OPEN) &&
                    !it_open_error(DISP_OPEN_OPEN, it)) {
                        /*
                         * If this is a successful OPEN request, we need to set
                         * replay handler and data early, so that if replay
                         * happens immediately after swabbing below, new reply
                         * is swabbed by that handler correctly.
                         */
                        mdc_set_open_replay_data(NULL, NULL, it);
                }

                if ((body->mbo_valid & (OBD_MD_FLDIREA | OBD_MD_FLEASIZE)) != 0) {
                        void *eadata;

                        mdc_update_max_ea_from_body(exp, body);

                        /*
                         * The eadata is opaque; just check that it is there.
                         * Eventually, obd_unpackmd() will check the contents.
                         */
                        eadata = req_capsule_server_sized_get(pill, &RMF_MDT_MD,
                                                              body->mbo_eadatasize);
                        if (!eadata)
                                return -EPROTO;

                        /* save lvb data and length in case this is for layout
                         * lock
                         */
                        lvb_data = eadata;
                        lvb_len = body->mbo_eadatasize;

                        /*
                         * We save the reply LOV EA in case we have to replay a
                         * create for recovery.  If we didn't allocate a large
                         * enough request buffer above we need to reallocate it
                         * here to hold the actual LOV EA.
                         *
                         * To not save LOV EA if request is not going to replay
                         * (for example error one).
                         */
                        if ((it->it_op & IT_OPEN) && req->rq_replay) {
                                void *lmm;

                                if (req_capsule_get_size(pill, &RMF_EADATA,
                                                         RCL_CLIENT) <
                                    body->mbo_eadatasize)
                                        mdc_realloc_openmsg(req, body);
                                else
                                        req_capsule_shrink(pill, &RMF_EADATA,
                                                           body->mbo_eadatasize,
                                                           RCL_CLIENT);

                                req_capsule_set_size(pill, &RMF_EADATA,
                                                     RCL_CLIENT,
                                                     body->mbo_eadatasize);

                                lmm = req_capsule_client_get(pill, &RMF_EADATA);
                                if (lmm)
                                        memcpy(lmm, eadata, body->mbo_eadatasize);
                        }
                }
        } else if (it->it_op & IT_LAYOUT) {
                /* maybe the lock was granted right away and layout
                 * is packed into RMF_DLM_LVB of req
                 */
                lvb_len = req_capsule_get_size(pill, &RMF_DLM_LVB, RCL_SERVER);
                if (lvb_len > 0) {
                        lvb_data = req_capsule_server_sized_get(pill,
                                                                &RMF_DLM_LVB,
                                                                lvb_len);
                        if (!lvb_data)
                                return -EPROTO;
                }
        }

        /* fill in stripe data for layout lock */
        lock = ldlm_handle2lock(lockh);
        if (lock && ldlm_has_layout(lock) && lvb_data) {
                void *lmm;

                LDLM_DEBUG(lock, "layout lock returned by: %s, lvb_len: %d",
                           ldlm_it2str(it->it_op), lvb_len);

                lmm = libcfs_kvzalloc(lvb_len, GFP_NOFS);
                if (!lmm) {
                        LDLM_LOCK_PUT(lock);
                        return -ENOMEM;
                }
                memcpy(lmm, lvb_data, lvb_len);

                /* install lvb_data */
                lock_res_and_lock(lock);
                if (!lock->l_lvb_data) {
                        lock->l_lvb_type = LVB_T_LAYOUT;
                        lock->l_lvb_data = lmm;
                        lock->l_lvb_len = lvb_len;
                        lmm = NULL;
                }
                unlock_res_and_lock(lock);
                if (lmm)
                        kvfree(lmm);
        }
        if (lock)
                LDLM_LOCK_PUT(lock);

        return rc;
}

/* We always reserve enough space in the reply packet for a stripe MD, because
 * we don't know in advance the file type.
 */
int mdc_enqueue(struct obd_export *exp, struct ldlm_enqueue_info *einfo,
                const union ldlm_policy_data *policy,
                struct lookup_intent *it, struct md_op_data *op_data,
                struct lustre_handle *lockh, u64 extra_lock_flags)
{
        static const union ldlm_policy_data lookup_policy = {
                .l_inodebits = { MDS_INODELOCK_LOOKUP }
        };
        static const union ldlm_policy_data update_policy = {
                .l_inodebits = { MDS_INODELOCK_UPDATE }
        };
        static const union ldlm_policy_data layout_policy = {
                .l_inodebits = { MDS_INODELOCK_LAYOUT }
        };
        static const union ldlm_policy_data getxattr_policy = {
                .l_inodebits = { MDS_INODELOCK_XATTR }
        };
        struct obd_device *obddev = class_exp2obd(exp);
        struct ptlrpc_request *req = NULL;
        u64 flags, saved_flags = extra_lock_flags;
        struct ldlm_res_id res_id;
        int generation, resends = 0;
        struct ldlm_reply *lockrep;
        enum lvb_type lvb_type = LVB_T_NONE;
        int rc;

        LASSERTF(!it || einfo->ei_type == LDLM_IBITS, "lock type %d\n",
                 einfo->ei_type);
        fid_build_reg_res_name(&op_data->op_fid1, &res_id);

        if (it) {
                LASSERT(!policy);

                saved_flags |= LDLM_FL_HAS_INTENT;
                if (it->it_op & (IT_UNLINK | IT_GETATTR | IT_READDIR))
                        policy = &update_policy;
                else if (it->it_op & IT_LAYOUT)
                        policy = &layout_policy;
                else if (it->it_op & (IT_GETXATTR | IT_SETXATTR))
                        policy = &getxattr_policy;
                else
                        policy = &lookup_policy;
        }

        generation = obddev->u.cli.cl_import->imp_generation;
resend:
        flags = saved_flags;
        if (!it) {
                /* The only way right now is FLOCK. */
                LASSERTF(einfo->ei_type == LDLM_FLOCK, "lock type %d\n",
                         einfo->ei_type);
                res_id.name[3] = LDLM_FLOCK;
        } else if (it->it_op & IT_OPEN) {
                req = mdc_intent_open_pack(exp, it, op_data);
        } else if (it->it_op & IT_UNLINK) {
                req = mdc_intent_unlink_pack(exp, it, op_data);
        } else if (it->it_op & (IT_GETATTR | IT_LOOKUP)) {
                req = mdc_intent_getattr_pack(exp, it, op_data);
        } else if (it->it_op & IT_READDIR) {
                req = mdc_enqueue_pack(exp, 0);
        } else if (it->it_op & IT_LAYOUT) {
                if (!imp_connect_lvb_type(class_exp2cliimp(exp)))
                        return -EOPNOTSUPP;
                req = mdc_intent_layout_pack(exp, it, op_data);
                lvb_type = LVB_T_LAYOUT;
        } else if (it->it_op & IT_GETXATTR) {
                req = mdc_intent_getxattr_pack(exp, it, op_data);
        } else {
                LBUG();
                return -EINVAL;
        }

        if (IS_ERR(req))
                return PTR_ERR(req);

        if (resends) {
                req->rq_generation_set = 1;
                req->rq_import_generation = generation;
                req->rq_sent = ktime_get_real_seconds() + resends;
        }

        /* It is important to obtain modify RPC slot first (if applicable), so
         * that threads that are waiting for a modify RPC slot are not polluting
         * our rpcs in flight counter.
         * We do not do flock request limiting, though
         */
        if (it) {
                mdc_get_mod_rpc_slot(req, it);
                rc = obd_get_request_slot(&obddev->u.cli);
                if (rc != 0) {
                        mdc_put_mod_rpc_slot(req, it);
                        mdc_clear_replay_flag(req, 0);
                        ptlrpc_req_finished(req);
                        return rc;
                }
        }

        rc = ldlm_cli_enqueue(exp, &req, einfo, &res_id, policy, &flags, NULL,
                              0, lvb_type, lockh, 0);
        if (!it) {
                /* For flock requests we immediately return without further
                 * delay and let caller deal with the rest, since rest of
                 * this function metadata processing makes no sense for flock
                 * requests anyway. But in case of problem during comms with
                 * Server (ETIMEDOUT) or any signal/kill attempt (EINTR), we
                 * can not rely on caller and this mainly for F_UNLCKs
                 * (explicits or automatically generated by Kernel to clean
                 * current FLocks upon exit) that can't be trashed
                 */
                if (((rc == -EINTR) || (rc == -ETIMEDOUT)) &&
                    (einfo->ei_type == LDLM_FLOCK) &&
                    (einfo->ei_mode == LCK_NL))
                        goto resend;
                return rc;
        }

        obd_put_request_slot(&obddev->u.cli);
        mdc_put_mod_rpc_slot(req, it);

        if (rc < 0) {
                CDEBUG(D_INFO, "%s: ldlm_cli_enqueue failed: rc = %d\n",
                       obddev->obd_name, rc);

                mdc_clear_replay_flag(req, rc);
                ptlrpc_req_finished(req);
                return rc;
        }

        lockrep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);

        lockrep->lock_policy_res2 =
                ptlrpc_status_ntoh(lockrep->lock_policy_res2);

        /*
         * Retry infinitely when the server returns -EINPROGRESS for the
         * intent operation, when server returns -EINPROGRESS for acquiring
         * intent lock, we'll retry in after_reply().
         */
        if (it->it_op && (int)lockrep->lock_policy_res2 == -EINPROGRESS) {
                mdc_clear_replay_flag(req, rc);
                ptlrpc_req_finished(req);
                resends++;

                CDEBUG(D_HA, "%s: resend:%d op:%d "DFID"/"DFID"\n",
                       obddev->obd_name, resends, it->it_op,
                       PFID(&op_data->op_fid1), PFID(&op_data->op_fid2));

                if (generation == obddev->u.cli.cl_import->imp_generation) {
                        goto resend;
                } else {
                        CDEBUG(D_HA, "resend cross eviction\n");
                        return -EIO;
                }
        }

        rc = mdc_finish_enqueue(exp, req, einfo, it, lockh, rc);
        if (rc < 0) {
                if (lustre_handle_is_used(lockh)) {
                        ldlm_lock_decref(lockh, einfo->ei_mode);
                        memset(lockh, 0, sizeof(*lockh));
                }
                ptlrpc_req_finished(req);

                it->it_lock_handle = 0;
                it->it_lock_mode = 0;
                it->it_request = NULL;
        }

        return rc;
}

static int mdc_finish_intent_lock(struct obd_export *exp,
                                  struct ptlrpc_request *request,
                                  struct md_op_data *op_data,
                                  struct lookup_intent *it,
                                  struct lustre_handle *lockh)
{
        struct lustre_handle old_lock;
        struct mdt_body *mdt_body;
        struct ldlm_lock *lock;
        int rc;

        LASSERT(request != LP_POISON);
        LASSERT(request->rq_repmsg != LP_POISON);

        if (it->it_op & IT_READDIR)
                return 0;

        if (!it_disposition(it, DISP_IT_EXECD)) {
                /* The server failed before it even started executing the
                 * intent, i.e. because it couldn't unpack the request.
                 */
                LASSERT(it->it_status != 0);
                return it->it_status;
        }
        rc = it_open_error(DISP_IT_EXECD, it);
        if (rc)
                return rc;

        mdt_body = req_capsule_server_get(&request->rq_pill, &RMF_MDT_BODY);
        LASSERT(mdt_body);      /* mdc_enqueue checked */

        rc = it_open_error(DISP_LOOKUP_EXECD, it);
        if (rc)
                return rc;

        /* keep requests around for the multiple phases of the call
         * this shows the DISP_XX must guarantee we make it into the call
         */
        if (!it_disposition(it, DISP_ENQ_CREATE_REF) &&
            it_disposition(it, DISP_OPEN_CREATE) &&
            !it_open_error(DISP_OPEN_CREATE, it)) {
                it_set_disposition(it, DISP_ENQ_CREATE_REF);
                ptlrpc_request_addref(request); /* balanced in ll_create_node */
        }
        if (!it_disposition(it, DISP_ENQ_OPEN_REF) &&
            it_disposition(it, DISP_OPEN_OPEN) &&
            !it_open_error(DISP_OPEN_OPEN, it)) {
                it_set_disposition(it, DISP_ENQ_OPEN_REF);
                ptlrpc_request_addref(request); /* balanced in ll_file_open */
                /* BUG 11546 - eviction in the middle of open rpc processing */
                OBD_FAIL_TIMEOUT(OBD_FAIL_MDC_ENQUEUE_PAUSE, obd_timeout);
        }

        if (it->it_op & IT_CREAT) {
                /* XXX this belongs in ll_create_it */
        } else if (it->it_op == IT_OPEN) {
                LASSERT(!it_disposition(it, DISP_OPEN_CREATE));
        } else {
                LASSERT(it->it_op & (IT_GETATTR | IT_LOOKUP | IT_LAYOUT));
        }

        /* If we already have a matching lock, then cancel the new
         * one.  We have to set the data here instead of in
         * mdc_enqueue, because we need to use the child's inode as
         * the l_ast_data to match, and that's not available until
         * intent_finish has performed the iget().)
         */
        lock = ldlm_handle2lock(lockh);
        if (lock) {
                union ldlm_policy_data policy = lock->l_policy_data;

                LDLM_DEBUG(lock, "matching against this");

                LASSERTF(fid_res_name_eq(&mdt_body->mbo_fid1,
                                         &lock->l_resource->lr_name),
                         "Lock res_id: "DLDLMRES", fid: "DFID"\n",
                         PLDLMRES(lock->l_resource), PFID(&mdt_body->mbo_fid1));
                LDLM_LOCK_PUT(lock);

                memcpy(&old_lock, lockh, sizeof(*lockh));
                if (ldlm_lock_match(NULL, LDLM_FL_BLOCK_GRANTED, NULL,
                                    LDLM_IBITS, &policy, LCK_NL,
                                    &old_lock, 0)) {
                        ldlm_lock_decref_and_cancel(lockh,
                                                    it->it_lock_mode);
                        memcpy(lockh, &old_lock, sizeof(old_lock));
                        it->it_lock_handle = lockh->cookie;
                }
        }
        CDEBUG(D_DENTRY,
               "D_IT dentry %.*s intent: %s status %d disp %x rc %d\n",
               (int)op_data->op_namelen, op_data->op_name,
               ldlm_it2str(it->it_op), it->it_status, it->it_disposition, rc);
        return rc;
}

int mdc_revalidate_lock(struct obd_export *exp, struct lookup_intent *it,
                        struct lu_fid *fid, __u64 *bits)
{
        /* We could just return 1 immediately, but since we should only
         * be called in revalidate_it if we already have a lock, let's
         * verify that.
         */
        struct ldlm_res_id res_id;
        struct lustre_handle lockh;
        union ldlm_policy_data policy;
        enum ldlm_mode mode;

        if (it->it_lock_handle) {
                lockh.cookie = it->it_lock_handle;
                mode = ldlm_revalidate_lock_handle(&lockh, bits);
        } else {
                fid_build_reg_res_name(fid, &res_id);
                switch (it->it_op) {
                case IT_GETATTR:
                        /* File attributes are held under multiple bits:
                         * nlink is under lookup lock, size and times are
                         * under UPDATE lock and recently we've also got
                         * a separate permissions lock for owner/group/acl that
                         * were protected by lookup lock before.
                         * Getattr must provide all of that information,
                         * so we need to ensure we have all of those locks.
                         * Unfortunately, if the bits are split across multiple
                         * locks, there's no easy way to match all of them here,
                         * so an extra RPC would be performed to fetch all
                         * of those bits at once for now.
                         */
                        /* For new MDTs(> 2.4), UPDATE|PERM should be enough,
                         * but for old MDTs (< 2.4), permission is covered
                         * by LOOKUP lock, so it needs to match all bits here.
                         */
                        policy.l_inodebits.bits = MDS_INODELOCK_UPDATE |
                                                  MDS_INODELOCK_LOOKUP |
                                                  MDS_INODELOCK_PERM;
                        break;
                case IT_READDIR:
                        policy.l_inodebits.bits = MDS_INODELOCK_UPDATE;
                        break;
                case IT_LAYOUT:
                        policy.l_inodebits.bits = MDS_INODELOCK_LAYOUT;

                        break;
                default:
                        policy.l_inodebits.bits = MDS_INODELOCK_LOOKUP;
                        break;
                }

                mode = mdc_lock_match(exp, LDLM_FL_BLOCK_GRANTED, fid,
                                      LDLM_IBITS, &policy,
                                      LCK_CR | LCK_CW | LCK_PR | LCK_PW,
                                      &lockh);
        }

        if (mode) {
                it->it_lock_handle = lockh.cookie;
                it->it_lock_mode = mode;
        } else {
                it->it_lock_handle = 0;
                it->it_lock_mode = 0;
        }

        return !!mode;
}

/*
 * This long block is all about fixing up the lock and request state
 * so that it is correct as of the moment _before_ the operation was
 * applied; that way, the VFS will think that everything is normal and
 * call Lustre's regular VFS methods.
 *
 * If we're performing a creation, that means that unless the creation
 * failed with EEXIST, we should fake up a negative dentry.
 *
 * For everything else, we want to lookup to succeed.
 *
 * One additional note: if CREATE or OPEN succeeded, we add an extra
 * reference to the request because we need to keep it around until
 * ll_create/ll_open gets called.
 *
 * The server will return to us, in it_disposition, an indication of
 * exactly what it_status refers to.
 *
 * If DISP_OPEN_OPEN is set, then it_status refers to the open() call,
 * otherwise if DISP_OPEN_CREATE is set, then it status is the
 * creation failure mode.  In either case, one of DISP_LOOKUP_NEG or
 * DISP_LOOKUP_POS will be set, indicating whether the child lookup
 * was successful.
 *
 * Else, if DISP_LOOKUP_EXECD then it_status is the rc of the
 * child lookup.
 */
int mdc_intent_lock(struct obd_export *exp, struct md_op_data *op_data,
                    struct lookup_intent *it, struct ptlrpc_request **reqp,
                    ldlm_blocking_callback cb_blocking, __u64 extra_lock_flags)
{
        struct ldlm_enqueue_info einfo = {
                .ei_type        = LDLM_IBITS,
                .ei_mode        = it_to_lock_mode(it),
                .ei_cb_bl       = cb_blocking,
                .ei_cb_cp       = ldlm_completion_ast,
        };
        struct lustre_handle lockh;
        int rc = 0;

        LASSERT(it);

        CDEBUG(D_DLMTRACE, "(name: %.*s,"DFID") in obj "DFID
                ", intent: %s flags %#Lo\n", (int)op_data->op_namelen,
                op_data->op_name, PFID(&op_data->op_fid2),
                PFID(&op_data->op_fid1), ldlm_it2str(it->it_op),
                it->it_flags);

        lockh.cookie = 0;
        if (fid_is_sane(&op_data->op_fid2) &&
            (it->it_op & (IT_LOOKUP | IT_GETATTR | IT_READDIR))) {
                /* We could just return 1 immediately, but since we should only
                 * be called in revalidate_it if we already have a lock, let's
                 * verify that.
                 */
                it->it_lock_handle = 0;
                rc = mdc_revalidate_lock(exp, it, &op_data->op_fid2, NULL);
                /* Only return failure if it was not GETATTR by cfid
                 * (from inode_revalidate)
                 */
                if (rc || op_data->op_namelen != 0)
                        return rc;
        }

        /* For case if upper layer did not alloc fid, do it now. */
        if (!fid_is_sane(&op_data->op_fid2) && it->it_op & IT_CREAT) {
                rc = mdc_fid_alloc(NULL, exp, &op_data->op_fid2, op_data);
                if (rc < 0) {
                        CERROR("Can't alloc new fid, rc %d\n", rc);
                        return rc;
                }
        }
        rc = mdc_enqueue(exp, &einfo, NULL, it, op_data, &lockh,
                         extra_lock_flags);
        if (rc < 0)
                return rc;

        *reqp = it->it_request;
        rc = mdc_finish_intent_lock(exp, *reqp, op_data, it, &lockh);
        return rc;
}

static int mdc_intent_getattr_async_interpret(const struct lu_env *env,
                                              struct ptlrpc_request *req,
                                              void *args, int rc)
{
        struct mdc_getattr_args  *ga = args;
        struct obd_export       *exp = ga->ga_exp;
        struct md_enqueue_info   *minfo = ga->ga_minfo;
        struct ldlm_enqueue_info *einfo = &minfo->mi_einfo;
        struct lookup_intent     *it;
        struct lustre_handle     *lockh;
        struct obd_device       *obddev;
        struct ldlm_reply        *lockrep;
        __u64                flags = LDLM_FL_HAS_INTENT;

        it    = &minfo->mi_it;
        lockh = &minfo->mi_lockh;

        obddev = class_exp2obd(exp);

        obd_put_request_slot(&obddev->u.cli);
        if (OBD_FAIL_CHECK(OBD_FAIL_MDC_GETATTR_ENQUEUE))
                rc = -ETIMEDOUT;

        rc = ldlm_cli_enqueue_fini(exp, req, einfo->ei_type, 1, einfo->ei_mode,
                                   &flags, NULL, 0, lockh, rc);
        if (rc < 0) {
                CERROR("ldlm_cli_enqueue_fini: %d\n", rc);
                mdc_clear_replay_flag(req, rc);
                goto out;
        }

        lockrep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);

        lockrep->lock_policy_res2 =
                ptlrpc_status_ntoh(lockrep->lock_policy_res2);

        rc = mdc_finish_enqueue(exp, req, einfo, it, lockh, rc);
        if (rc)
                goto out;

        rc = mdc_finish_intent_lock(exp, req, &minfo->mi_data, it, lockh);

out:
        minfo->mi_cb(req, minfo, rc);
        return 0;
}

int mdc_intent_getattr_async(struct obd_export *exp,
                             struct md_enqueue_info *minfo)
{
        struct md_op_data       *op_data = &minfo->mi_data;
        struct lookup_intent    *it = &minfo->mi_it;
        struct ptlrpc_request   *req;
        struct mdc_getattr_args *ga;
        struct obd_device       *obddev = class_exp2obd(exp);
        struct ldlm_res_id       res_id;
        union ldlm_policy_data policy = {
                .l_inodebits = { MDS_INODELOCK_LOOKUP | MDS_INODELOCK_UPDATE }
        };
        int                   rc = 0;
        __u64               flags = LDLM_FL_HAS_INTENT;

        CDEBUG(D_DLMTRACE,
               "name: %.*s in inode " DFID ", intent: %s flags %#Lo\n",
               (int)op_data->op_namelen, op_data->op_name,
               PFID(&op_data->op_fid1), ldlm_it2str(it->it_op), it->it_flags);

        fid_build_reg_res_name(&op_data->op_fid1, &res_id);
        req = mdc_intent_getattr_pack(exp, it, op_data);
        if (IS_ERR(req))
                return PTR_ERR(req);

        rc = obd_get_request_slot(&obddev->u.cli);
        if (rc != 0) {
                ptlrpc_req_finished(req);
                return rc;
        }

        rc = ldlm_cli_enqueue(exp, &req, &minfo->mi_einfo, &res_id, &policy,
                              &flags, NULL, 0, LVB_T_NONE, &minfo->mi_lockh, 1);
        if (rc < 0) {
                obd_put_request_slot(&obddev->u.cli);
                ptlrpc_req_finished(req);
                return rc;
        }

        BUILD_BUG_ON(sizeof(*ga) > sizeof(req->rq_async_args));
        ga = ptlrpc_req_async_args(req);
        ga->ga_exp = exp;
        ga->ga_minfo = minfo;

        req->rq_interpret_reply = mdc_intent_getattr_async_interpret;
        ptlrpcd_add_req(req);

        return 0;
}