lustre/mdc/mdc_locks.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2014, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  */
  36
  37 #define DEBUG_SUBSYSTEM S_MDC
  38
  39 #include <linux/module.h>
  40 #include <obd.h>
  41 #include <obd_class.h>
  42 #include <lustre_dlm.h>
  43 #include <lustre_fid.h> /* fid_res_name_eq() */
  44 #include <lustre_intent.h>
  45 #include <lustre_mdc.h>
  46 #include <lustre_net.h>
  47 #include <lustre_req_layout.h>
  48 #include "mdc_internal.h"
  49
  50 struct mdc_getattr_args {
  51         struct obd_export           *ga_exp;
  52         struct md_enqueue_info      *ga_minfo;
  53         struct ldlm_enqueue_info    *ga_einfo;
  54 };
  55
  56 int it_open_error(int phase, struct lookup_intent *it)
  57 {
  58         if (it_disposition(it, DISP_OPEN_LEASE)) {
  59                 if (phase >= DISP_OPEN_LEASE)
  60                         return it->d.lustre.it_status;
  61                 else
  62                         return 0;
  63         }
  64         if (it_disposition(it, DISP_OPEN_OPEN)) {
  65                 if (phase >= DISP_OPEN_OPEN)
  66                         return it->d.lustre.it_status;
  67                 else
  68                         return 0;
  69         }
  70
  71         if (it_disposition(it, DISP_OPEN_CREATE)) {
  72                 if (phase >= DISP_OPEN_CREATE)
  73                         return it->d.lustre.it_status;
  74                 else
  75                         return 0;
  76         }
  77
  78         if (it_disposition(it, DISP_LOOKUP_EXECD)) {
  79                 if (phase >= DISP_LOOKUP_EXECD)
  80                         return it->d.lustre.it_status;
  81                 else
  82                         return 0;
  83         }
  84
  85         if (it_disposition(it, DISP_IT_EXECD)) {
  86                 if (phase >= DISP_IT_EXECD)
  87                         return it->d.lustre.it_status;
  88                 else
  89                         return 0;
  90         }
  91         CERROR("it disp: %X, status: %d\n", it->d.lustre.it_disposition,
  92                it->d.lustre.it_status);
  93         LBUG();
  94         return 0;
  95 }
  96 EXPORT_SYMBOL(it_open_error);
  97
  98 /* this must be called on a lockh that is known to have a referenced lock */
  99 int mdc_set_lock_data(struct obd_export *exp, __u64 *lockh, void *data,
 100                       __u64 *bits)
 101 {
 102         struct ldlm_lock *lock;
 103         struct inode *new_inode = data;
 104         ENTRY;
 105
 106         if(bits)
 107                 *bits = 0;
 108
 109         if (!*lockh)
 110                 RETURN(0);
 111
 112         lock = ldlm_handle2lock((struct lustre_handle *)lockh);
 113
 114         LASSERT(lock != NULL);
 115         lock_res_and_lock(lock);
 116         if (lock->l_resource->lr_lvb_inode &&
 117             lock->l_resource->lr_lvb_inode != data) {
 118                 struct inode *old_inode = lock->l_resource->lr_lvb_inode;
 119                 LASSERTF(old_inode->i_state & I_FREEING,
 120                          "Found existing inode %p/%lu/%u state %lu in lock: "
 121                          "setting data to %p/%lu/%u\n", old_inode,
 122                          old_inode->i_ino, old_inode->i_generation,
 123                          old_inode->i_state,
 124                          new_inode, new_inode->i_ino, new_inode->i_generation);
 125         }
 126         lock->l_resource->lr_lvb_inode = new_inode;
 127         if (bits)
 128                 *bits = lock->l_policy_data.l_inodebits.bits;
 129
 130         unlock_res_and_lock(lock);
 131         LDLM_LOCK_PUT(lock);
 132
 133         RETURN(0);
 134 }
 135
 136 ldlm_mode_t mdc_lock_match(struct obd_export *exp, __u64 flags,
 137                            const struct lu_fid *fid, ldlm_type_t type,
 138                            ldlm_policy_data_t *policy, ldlm_mode_t mode,
 139                            struct lustre_handle *lockh)
 140 {
 141         struct ldlm_res_id res_id;
 142         ldlm_mode_t rc;
 143         ENTRY;
 144
 145         fid_build_reg_res_name(fid, &res_id);
 146         /* LU-4405: Clear bits not supported by server */
 147         policy->l_inodebits.bits &= exp_connect_ibits(exp);
 148         rc = ldlm_lock_match(class_exp2obd(exp)->obd_namespace, flags,
 149                              &res_id, type, policy, mode, lockh, 0);
 150         RETURN(rc);
 151 }
 152
 153 int mdc_cancel_unused(struct obd_export *exp,
 154                       const struct lu_fid *fid,
 155                       ldlm_policy_data_t *policy,
 156                       ldlm_mode_t mode,
 157                       ldlm_cancel_flags_t flags,
 158                       void *opaque)
 159 {
 160         struct ldlm_res_id res_id;
 161         struct obd_device *obd = class_exp2obd(exp);
 162         int rc;
 163
 164         ENTRY;
 165
 166         fid_build_reg_res_name(fid, &res_id);
 167         rc = ldlm_cli_cancel_unused_resource(obd->obd_namespace, &res_id,
 168                                              policy, mode, flags, opaque);
 169         RETURN(rc);
 170 }
 171
 172 int mdc_null_inode(struct obd_export *exp,
 173                    const struct lu_fid *fid)
 174 {
 175         struct ldlm_res_id res_id;
 176         struct ldlm_resource *res;
 177         struct ldlm_namespace *ns = class_exp2obd(exp)->obd_namespace;
 178         ENTRY;
 179
 180         LASSERTF(ns != NULL, "no namespace passed\n");
 181
 182         fid_build_reg_res_name(fid, &res_id);
 183
 184         res = ldlm_resource_get(ns, NULL, &res_id, 0, 0);
 185         if (IS_ERR(res))
 186                 RETURN(0);
 187
 188         lock_res(res);
 189         res->lr_lvb_inode = NULL;
 190         unlock_res(res);
 191
 192         ldlm_resource_putref(res);
 193         RETURN(0);
 194 }
 195
 196 /* find any ldlm lock of the inode in mdc
 197  * return 0    not find
 198  *        1    find one
 199  *      < 0    error */
 200 int mdc_find_cbdata(struct obd_export *exp,
 201                     const struct lu_fid *fid,
 202                     ldlm_iterator_t it, void *data)
 203 {
 204         struct ldlm_res_id res_id;
 205         int rc = 0;
 206         ENTRY;
 207
 208         fid_build_reg_res_name((struct lu_fid*)fid, &res_id);
 209         rc = ldlm_resource_iterate(class_exp2obd(exp)->obd_namespace, &res_id,
 210                                    it, data);
 211         if (rc == LDLM_ITER_STOP)
 212                 RETURN(1);
 213         else if (rc == LDLM_ITER_CONTINUE)
 214                 RETURN(0);
 215         RETURN(rc);
 216 }
 217
 218 static inline void mdc_clear_replay_flag(struct ptlrpc_request *req, int rc)
 219 {
 220         /* Don't hold error requests for replay. */
 221         if (req->rq_replay) {
 222                 spin_lock(&req->rq_lock);
 223                 req->rq_replay = 0;
 224                 spin_unlock(&req->rq_lock);
 225         }
 226         if (rc && req->rq_transno != 0) {
 227                 DEBUG_REQ(D_ERROR, req, "transno returned on error rc %d", rc);
 228                 LBUG();
 229         }
 230 }
 231
 232 /* Save a large LOV EA into the request buffer so that it is available
 233  * for replay.  We don't do this in the initial request because the
 234  * original request doesn't need this buffer (at most it sends just the
 235  * lov_mds_md) and it is a waste of RAM/bandwidth to send the empty
 236  * buffer and may also be difficult to allocate and save a very large
 237  * request buffer for each open. (bug 5707)
 238  *
 239  * OOM here may cause recovery failure if lmm is needed (only for the
 240  * original open if the MDS crashed just when this client also OOM'd)
 241  * but this is incredibly unlikely, and questionable whether the client
 242  * could do MDS recovery under OOM anyways... */
 243 static void mdc_realloc_openmsg(struct ptlrpc_request *req,
 244                                 struct mdt_body *body)
 245 {
 246         int     rc;
 247
 248         /* FIXME: remove this explicit offset. */
 249         rc = sptlrpc_cli_enlarge_reqbuf(req, DLM_INTENT_REC_OFF + 4,
 250                                         body->mbo_eadatasize);
 251         if (rc) {
 252                 CERROR("Can't enlarge segment %d size to %d\n",
 253                        DLM_INTENT_REC_OFF + 4, body->mbo_eadatasize);
 254                 body->mbo_valid &= ~OBD_MD_FLEASIZE;
 255                 body->mbo_eadatasize = 0;
 256         }
 257 }
 258
 259 static struct ptlrpc_request *
 260 mdc_intent_open_pack(struct obd_export *exp, struct lookup_intent *it,
 261                      struct md_op_data *op_data)
 262 {
 263         struct ptlrpc_request   *req;
 264         struct obd_device       *obddev = class_exp2obd(exp);
 265         struct ldlm_intent      *lit;
 266         const void              *lmm = op_data->op_data;
 267         __u32                    lmmsize = op_data->op_data_size;
 268         struct list_head         cancels = LIST_HEAD_INIT(cancels);
 269         int                      count = 0;
 270         int                      mode;
 271         int                      rc;
 272         ENTRY;
 273
 274         it->it_create_mode = (it->it_create_mode & ~S_IFMT) | S_IFREG;
 275
 276         /* XXX: openlock is not cancelled for cross-refs. */
 277         /* If inode is known, cancel conflicting OPEN locks. */
 278         if (fid_is_sane(&op_data->op_fid2)) {
 279                 if (it->it_flags & MDS_OPEN_LEASE) { /* try to get lease */
 280                         if (it->it_flags & FMODE_WRITE)
 281                                 mode = LCK_EX;
 282                         else
 283                                 mode = LCK_PR;
 284                 } else {
 285                         if (it->it_flags & (FMODE_WRITE|MDS_OPEN_TRUNC))
 286                                 mode = LCK_CW;
 287 #ifdef FMODE_EXEC
 288                         else if (it->it_flags & FMODE_EXEC)
 289                                 mode = LCK_PR;
 290 #endif
 291                         else
 292                                 mode = LCK_CR;
 293                 }
 294                 count = mdc_resource_get_unused(exp, &op_data->op_fid2,
 295                                                 &cancels, mode,
 296                                                 MDS_INODELOCK_OPEN);
 297         }
 298
 299         /* If CREATE, cancel parent's UPDATE lock. */
 300         if (it->it_op & IT_CREAT)
 301                 mode = LCK_EX;
 302         else
 303                 mode = LCK_CR;
 304         count += mdc_resource_get_unused(exp, &op_data->op_fid1,
 305                                          &cancels, mode,
 306                                          MDS_INODELOCK_UPDATE);
 307
 308         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 309                                    &RQF_LDLM_INTENT_OPEN);
 310         if (req == NULL) {
 311                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
 312                 RETURN(ERR_PTR(-ENOMEM));
 313         }
 314
 315         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
 316                              op_data->op_namelen + 1);
 317         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT,
 318                              max(lmmsize, obddev->u.cli.cl_default_mds_easize));
 319
 320         rc = ldlm_prep_enqueue_req(exp, req, &cancels, count);
 321         if (rc < 0) {
 322                 ptlrpc_request_free(req);
 323                 RETURN(ERR_PTR(rc));
 324         }
 325
 326         spin_lock(&req->rq_lock);
 327         req->rq_replay = req->rq_import->imp_replayable;
 328         spin_unlock(&req->rq_lock);
 329
 330         /* pack the intent */
 331         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
 332         lit->opc = (__u64)it->it_op;
 333
 334         /* pack the intended request */
 335         mdc_open_pack(req, op_data, it->it_create_mode, 0, it->it_flags, lmm,
 336                       lmmsize);
 337
 338         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
 339                              obddev->u.cli.cl_max_mds_easize);
 340
 341         /* for remote client, fetch remote perm for current user */
 342         if (client_is_remote(exp))
 343                 req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER,
 344                                      sizeof(struct mdt_remote_perm));
 345         ptlrpc_request_set_replen(req);
 346         return req;
 347 }
 348
 349 static struct ptlrpc_request *
 350 mdc_intent_getxattr_pack(struct obd_export *exp,
 351                          struct lookup_intent *it,
 352                          struct md_op_data *op_data)
 353 {
 354         struct ptlrpc_request   *req;
 355         struct ldlm_intent      *lit;
 356         int                     rc, count = 0;
 357         __u32                   maxdata;
 358         struct list_head        cancels = LIST_HEAD_INIT(cancels);
 359
 360         ENTRY;
 361
 362         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 363                                         &RQF_LDLM_INTENT_GETXATTR);
 364         if (req == NULL)
 365                 RETURN(ERR_PTR(-ENOMEM));
 366
 367         rc = ldlm_prep_enqueue_req(exp, req, &cancels, count);
 368         if (rc) {
 369                 ptlrpc_request_free(req);
 370                 RETURN(ERR_PTR(rc));
 371         }
 372
 373         /* pack the intent */
 374         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
 375         lit->opc = IT_GETXATTR;
 376
 377         maxdata = class_exp2cliimp(exp)->imp_connect_data.ocd_max_easize;
 378
 379         /* pack the intended request */
 380         mdc_pack_body(req, &op_data->op_fid1, op_data->op_valid, maxdata, -1,
 381                       0);
 382
 383         req_capsule_set_size(&req->rq_pill, &RMF_EADATA,
 384                                 RCL_SERVER, maxdata);
 385
 386         req_capsule_set_size(&req->rq_pill, &RMF_EAVALS,
 387                                 RCL_SERVER, maxdata);
 388
 389         req_capsule_set_size(&req->rq_pill, &RMF_EAVALS_LENS,
 390                                 RCL_SERVER, maxdata);
 391
 392         ptlrpc_request_set_replen(req);
 393
 394         RETURN(req);
 395 }
 396
 397 static struct ptlrpc_request *mdc_intent_unlink_pack(struct obd_export *exp,
 398                                                      struct lookup_intent *it,
 399                                                      struct md_op_data *op_data)
 400 {
 401         struct ptlrpc_request *req;
 402         struct obd_device     *obddev = class_exp2obd(exp);
 403         struct ldlm_intent    *lit;
 404         int                    rc;
 405         ENTRY;
 406
 407         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 408                                    &RQF_LDLM_INTENT_UNLINK);
 409         if (req == NULL)
 410                 RETURN(ERR_PTR(-ENOMEM));
 411
 412         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
 413                              op_data->op_namelen + 1);
 414
 415         rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
 416         if (rc) {
 417                 ptlrpc_request_free(req);
 418                 RETURN(ERR_PTR(rc));
 419         }
 420
 421         /* pack the intent */
 422         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
 423         lit->opc = (__u64)it->it_op;
 424
 425         /* pack the intended request */
 426         mdc_unlink_pack(req, op_data);
 427
 428         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
 429                              obddev->u.cli.cl_default_mds_easize);
 430         ptlrpc_request_set_replen(req);
 431         RETURN(req);
 432 }
 433
 434 static struct ptlrpc_request *mdc_intent_getattr_pack(struct obd_export *exp,
 435                                                       struct lookup_intent *it,
 436                                                       struct md_op_data *op_data)
 437 {
 438         struct ptlrpc_request   *req;
 439         struct obd_device       *obddev = class_exp2obd(exp);
 440         u64                      valid = OBD_MD_FLGETATTR | OBD_MD_FLEASIZE |
 441                                          OBD_MD_FLMODEASIZE | OBD_MD_FLDIREA |
 442                                          OBD_MD_MEA |
 443                                          (client_is_remote(exp) ?
 444                                           OBD_MD_FLRMTPERM : OBD_MD_FLACL);
 445         struct ldlm_intent      *lit;
 446         int                      rc;
 447         __u32                    easize;
 448         ENTRY;
 449
 450         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 451                                    &RQF_LDLM_INTENT_GETATTR);
 452         if (req == NULL)
 453                 RETURN(ERR_PTR(-ENOMEM));
 454
 455         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
 456                              op_data->op_namelen + 1);
 457
 458         rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
 459         if (rc) {
 460                 ptlrpc_request_free(req);
 461                 RETURN(ERR_PTR(rc));
 462         }
 463
 464         /* pack the intent */
 465         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
 466         lit->opc = (__u64)it->it_op;
 467
 468         if (obddev->u.cli.cl_default_mds_easize > 0)
 469                 easize = obddev->u.cli.cl_default_mds_easize;
 470         else
 471                 easize = obddev->u.cli.cl_max_mds_easize;
 472
 473         /* pack the intended request */
 474         mdc_getattr_pack(req, valid, it->it_flags, op_data, easize);
 475
 476         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER, easize);
 477         if (client_is_remote(exp))
 478                 req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER,
 479                                      sizeof(struct mdt_remote_perm));
 480         ptlrpc_request_set_replen(req);
 481         RETURN(req);
 482 }
 483
 484 static struct ptlrpc_request *mdc_intent_layout_pack(struct obd_export *exp,
 485                                                      struct lookup_intent *it,
 486                                                      struct md_op_data *unused)
 487 {
 488         struct obd_device     *obd = class_exp2obd(exp);
 489         struct ptlrpc_request *req;
 490         struct ldlm_intent    *lit;
 491         struct layout_intent  *layout;
 492         int rc;
 493         ENTRY;
 494
 495         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 496                                 &RQF_LDLM_INTENT_LAYOUT);
 497         if (req == NULL)
 498                 RETURN(ERR_PTR(-ENOMEM));
 499
 500         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT, 0);
 501         rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
 502         if (rc) {
 503                 ptlrpc_request_free(req);
 504                 RETURN(ERR_PTR(rc));
 505         }
 506
 507         /* pack the intent */
 508         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
 509         lit->opc = (__u64)it->it_op;
 510
 511         /* pack the layout intent request */
 512         layout = req_capsule_client_get(&req->rq_pill, &RMF_LAYOUT_INTENT);
 513         /* LAYOUT_INTENT_ACCESS is generic, specific operation will be
 514          * set for replication */
 515         layout->li_opc = LAYOUT_INTENT_ACCESS;
 516
 517         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
 518                              obd->u.cli.cl_default_mds_easize);
 519         ptlrpc_request_set_replen(req);
 520         RETURN(req);
 521 }
 522
 523 static struct ptlrpc_request *
 524 mdc_enqueue_pack(struct obd_export *exp, int lvb_len)
 525 {
 526         struct ptlrpc_request *req;
 527         int rc;
 528         ENTRY;
 529
 530         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LDLM_ENQUEUE);
 531         if (req == NULL)
 532                 RETURN(ERR_PTR(-ENOMEM));
 533
 534         rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
 535         if (rc) {
 536                 ptlrpc_request_free(req);
 537                 RETURN(ERR_PTR(rc));
 538         }
 539
 540         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER, lvb_len);
 541         ptlrpc_request_set_replen(req);
 542         RETURN(req);
 543 }
 544
 545 static int mdc_finish_enqueue(struct obd_export *exp,
 546                               struct ptlrpc_request *req,
 547                               struct ldlm_enqueue_info *einfo,
 548                               struct lookup_intent *it,
 549                               struct lustre_handle *lockh,
 550                               int rc)
 551 {
 552         struct req_capsule  *pill = &req->rq_pill;
 553         struct ldlm_request *lockreq;
 554         struct ldlm_reply   *lockrep;
 555         struct lustre_intent_data *intent = &it->d.lustre;
 556         struct ldlm_lock    *lock;
 557         void                *lvb_data = NULL;
 558         __u32                lvb_len = 0;
 559         ENTRY;
 560
 561         LASSERT(rc >= 0);
 562         /* Similarly, if we're going to replay this request, we don't want to
 563          * actually get a lock, just perform the intent. */
 564         if (req->rq_transno || req->rq_replay) {
 565                 lockreq = req_capsule_client_get(pill, &RMF_DLM_REQ);
 566                 lockreq->lock_flags |= ldlm_flags_to_wire(LDLM_FL_INTENT_ONLY);
 567         }
 568
 569         if (rc == ELDLM_LOCK_ABORTED) {
 570                 einfo->ei_mode = 0;
 571                 memset(lockh, 0, sizeof(*lockh));
 572                 rc = 0;
 573         } else { /* rc = 0 */
 574                 lock = ldlm_handle2lock(lockh);
 575                 LASSERT(lock != NULL);
 576
 577                 /* If the server gave us back a different lock mode, we should
 578                  * fix up our variables. */
 579                 if (lock->l_req_mode != einfo->ei_mode) {
 580                         ldlm_lock_addref(lockh, lock->l_req_mode);
 581                         ldlm_lock_decref(lockh, einfo->ei_mode);
 582                         einfo->ei_mode = lock->l_req_mode;
 583                 }
 584                 LDLM_LOCK_PUT(lock);
 585         }
 586
 587         lockrep = req_capsule_server_get(pill, &RMF_DLM_REP);
 588         LASSERT(lockrep != NULL); /* checked by ldlm_cli_enqueue() */
 589
 590         intent->it_disposition = (int)lockrep->lock_policy_res1;
 591         intent->it_status = (int)lockrep->lock_policy_res2;
 592         intent->it_lock_mode = einfo->ei_mode;
 593         intent->it_lock_handle = lockh->cookie;
 594         intent->it_data = req;
 595
 596         /* Technically speaking rq_transno must already be zero if
 597          * it_status is in error, so the check is a bit redundant */
 598         if ((!req->rq_transno || intent->it_status < 0) && req->rq_replay)
 599                 mdc_clear_replay_flag(req, intent->it_status);
 600
 601         /* If we're doing an IT_OPEN which did not result in an actual
 602          * successful open, then we need to remove the bit which saves
 603          * this request for unconditional replay.
 604          *
 605          * It's important that we do this first!  Otherwise we might exit the
 606          * function without doing so, and try to replay a failed create
 607          * (bug 3440) */
 608         if (it->it_op & IT_OPEN && req->rq_replay &&
 609             (!it_disposition(it, DISP_OPEN_OPEN) ||intent->it_status != 0))
 610                 mdc_clear_replay_flag(req, intent->it_status);
 611
 612         DEBUG_REQ(D_RPCTRACE, req, "op: %d disposition: %x, status: %d",
 613                   it->it_op, intent->it_disposition, intent->it_status);
 614
 615         /* We know what to expect, so we do any byte flipping required here */
 616         if (it->it_op & (IT_OPEN | IT_UNLINK | IT_LOOKUP | IT_GETATTR)) {
 617                 struct mdt_body *body;
 618
 619                 body = req_capsule_server_get(pill, &RMF_MDT_BODY);
 620                 if (body == NULL) {
 621                         CERROR ("Can't swab mdt_body\n");
 622                         RETURN (-EPROTO);
 623                 }
 624
 625                 if (it_disposition(it, DISP_OPEN_OPEN) &&
 626                     !it_open_error(DISP_OPEN_OPEN, it)) {
 627                         /*
 628                          * If this is a successful OPEN request, we need to set
 629                          * replay handler and data early, so that if replay
 630                          * happens immediately after swabbing below, new reply
 631                          * is swabbed by that handler correctly.
 632                          */
 633                         mdc_set_open_replay_data(NULL, NULL, it);
 634                 }
 635
 636                 if (body->mbo_valid & (OBD_MD_FLDIREA | OBD_MD_FLEASIZE)) {
 637                         void *eadata;
 638
 639                         mdc_update_max_ea_from_body(exp, body);
 640
 641                         /*
 642                          * The eadata is opaque; just check that it is there.
 643                          * Eventually, obd_unpackmd() will check the contents.
 644                          */
 645                         eadata = req_capsule_server_sized_get(pill, &RMF_MDT_MD,
 646                                                         body->mbo_eadatasize);
 647                         if (eadata == NULL)
 648                                 RETURN(-EPROTO);
 649
 650                         /* save lvb data and length in case this is for layout
 651                          * lock */
 652                         lvb_data = eadata;
 653                         lvb_len = body->mbo_eadatasize;
 654
 655                         /*
 656                          * We save the reply LOV EA in case we have to replay a
 657                          * create for recovery.  If we didn't allocate a large
 658                          * enough request buffer above we need to reallocate it
 659                          * here to hold the actual LOV EA.
 660                          *
 661                          * To not save LOV EA if request is not going to replay
 662                          * (for example error one).
 663                          */
 664                         if ((it->it_op & IT_OPEN) && req->rq_replay) {
 665                                 void *lmm;
 666                                 if (req_capsule_get_size(pill, &RMF_EADATA,
 667                                                          RCL_CLIENT) <
 668                                     body->mbo_eadatasize)
 669                                         mdc_realloc_openmsg(req, body);
 670                                 else
 671                                         req_capsule_shrink(pill, &RMF_EADATA,
 672                                                            body->mbo_eadatasize,
 673                                                            RCL_CLIENT);
 674
 675                                 req_capsule_set_size(pill, &RMF_EADATA,
 676                                                      RCL_CLIENT,
 677                                                      body->mbo_eadatasize);
 678
 679                                 lmm = req_capsule_client_get(pill, &RMF_EADATA);
 680                                 if (lmm)
 681                                         memcpy(lmm, eadata,
 682                                                body->mbo_eadatasize);
 683                         }
 684                 }
 685
 686                 if (body->mbo_valid & OBD_MD_FLRMTPERM) {
 687                         struct mdt_remote_perm *perm;
 688
 689                         LASSERT(client_is_remote(exp));
 690                         perm = req_capsule_server_swab_get(pill, &RMF_ACL,
 691                                                 lustre_swab_mdt_remote_perm);
 692                         if (perm == NULL)
 693                                 RETURN(-EPROTO);
 694                 }
 695         } else if (it->it_op & IT_LAYOUT) {
 696                 /* maybe the lock was granted right away and layout
 697                  * is packed into RMF_DLM_LVB of req */
 698                 lvb_len = req_capsule_get_size(pill, &RMF_DLM_LVB, RCL_SERVER);
 699                 if (lvb_len > 0) {
 700                         lvb_data = req_capsule_server_sized_get(pill,
 701                                                         &RMF_DLM_LVB, lvb_len);
 702                         if (lvb_data == NULL)
 703                                 RETURN(-EPROTO);
 704                 }
 705         }
 706
 707         /* fill in stripe data for layout lock.
 708          * LU-6581: trust layout data only if layout lock is granted. The MDT
 709          * has stopped sending layout unless the layout lock is granted. The
 710          * client still does this checking in case it's talking with an old
 711          * server. - Jinshan */
 712         lock = ldlm_handle2lock(lockh);
 713         if (lock != NULL && ldlm_has_layout(lock) && lvb_data != NULL &&
 714             !(lockrep->lock_flags & LDLM_FL_BLOCKED_MASK)) {
 715                 void *lmm;
 716
 717                 LDLM_DEBUG(lock, "layout lock returned by: %s, lvb_len: %d\n",
 718                         ldlm_it2str(it->it_op), lvb_len);
 719
 720                 OBD_ALLOC_LARGE(lmm, lvb_len);
 721                 if (lmm == NULL) {
 722                         LDLM_LOCK_PUT(lock);
 723                         RETURN(-ENOMEM);
 724                 }
 725                 memcpy(lmm, lvb_data, lvb_len);
 726
 727                 /* install lvb_data */
 728                 lock_res_and_lock(lock);
 729                 if (lock->l_lvb_data == NULL) {
 730                         lock->l_lvb_type = LVB_T_LAYOUT;
 731                         lock->l_lvb_data = lmm;
 732                         lock->l_lvb_len = lvb_len;
 733                         lmm = NULL;
 734                 }
 735                 unlock_res_and_lock(lock);
 736                 if (lmm != NULL)
 737                         OBD_FREE_LARGE(lmm, lvb_len);
 738         }
 739         if (lock != NULL)
 740                 LDLM_LOCK_PUT(lock);
 741
 742         RETURN(rc);
 743 }
 744
 745 /* We always reserve enough space in the reply packet for a stripe MD, because
 746  * we don't know in advance the file type. */
 747 int mdc_enqueue(struct obd_export *exp,
 748                 struct ldlm_enqueue_info *einfo,
 749                 const union ldlm_policy_data *policy,
 750                 struct lookup_intent *it, struct md_op_data *op_data,
 751                 struct lustre_handle *lockh, __u64 extra_lock_flags)
 752 {
 753         struct obd_device     *obddev = class_exp2obd(exp);
 754         struct ptlrpc_request *req = NULL;
 755         __u64                  flags, saved_flags = extra_lock_flags;
 756         int                    rc;
 757         struct ldlm_res_id res_id;
 758         static const ldlm_policy_data_t lookup_policy =
 759                             { .l_inodebits = { MDS_INODELOCK_LOOKUP } };
 760         static const ldlm_policy_data_t update_policy =
 761                             { .l_inodebits = { MDS_INODELOCK_UPDATE } };
 762         static const ldlm_policy_data_t layout_policy =
 763                             { .l_inodebits = { MDS_INODELOCK_LAYOUT } };
 764         static const ldlm_policy_data_t getxattr_policy = {
 765                               .l_inodebits = { MDS_INODELOCK_XATTR } };
 766         int                    generation, resends = 0;
 767         struct ldlm_reply     *lockrep;
 768         enum lvb_type          lvb_type = 0;
 769         ENTRY;
 770
 771         LASSERTF(!it || einfo->ei_type == LDLM_IBITS, "lock type %d\n",
 772                  einfo->ei_type);
 773         fid_build_reg_res_name(&op_data->op_fid1, &res_id);
 774
 775         if (it != NULL) {
 776                 LASSERT(policy == NULL);
 777
 778                 saved_flags |= LDLM_FL_HAS_INTENT;
 779                 if (it->it_op & (IT_OPEN | IT_UNLINK | IT_GETATTR | IT_READDIR))
 780                         policy = &update_policy;
 781                 else if (it->it_op & IT_LAYOUT)
 782                         policy = &layout_policy;
 783                 else if (it->it_op & (IT_GETXATTR | IT_SETXATTR))
 784                         policy = &getxattr_policy;
 785                 else
 786                         policy = &lookup_policy;
 787         }
 788
 789         generation = obddev->u.cli.cl_import->imp_generation;
 790 resend:
 791         flags = saved_flags;
 792         if (it == NULL) {
 793                 /* The only way right now is FLOCK. */
 794                 LASSERTF(einfo->ei_type == LDLM_FLOCK, "lock type %d\n",
 795                          einfo->ei_type);
 796                 res_id.name[3] = LDLM_FLOCK;
 797         } else if (it->it_op & IT_OPEN) {
 798                 req = mdc_intent_open_pack(exp, it, op_data);
 799         } else if (it->it_op & IT_UNLINK) {
 800                 req = mdc_intent_unlink_pack(exp, it, op_data);
 801         } else if (it->it_op & (IT_GETATTR | IT_LOOKUP)) {
 802                 req = mdc_intent_getattr_pack(exp, it, op_data);
 803         } else if (it->it_op & IT_READDIR) {
 804                 req = mdc_enqueue_pack(exp, 0);
 805         } else if (it->it_op & IT_LAYOUT) {
 806                 if (!imp_connect_lvb_type(class_exp2cliimp(exp)))
 807                         RETURN(-EOPNOTSUPP);
 808                 req = mdc_intent_layout_pack(exp, it, op_data);
 809                 lvb_type = LVB_T_LAYOUT;
 810         } else if (it->it_op & IT_GETXATTR) {
 811                 req = mdc_intent_getxattr_pack(exp, it, op_data);
 812         } else {
 813                 LBUG();
 814                 RETURN(-EINVAL);
 815         }
 816
 817         if (IS_ERR(req))
 818                 RETURN(PTR_ERR(req));
 819
 820         if (resends) {
 821                 req->rq_generation_set = 1;
 822                 req->rq_import_generation = generation;
 823                 req->rq_sent = cfs_time_current_sec() + resends;
 824         }
 825
 826         /* It is important to obtain modify RPC slot first (if applicable), so
 827          * that threads that are waiting for a modify RPC slot are not polluting
 828          * our rpcs in flight counter.
 829          * We do not do flock request limiting, though */
 830         if (it) {
 831                 mdc_get_mod_rpc_slot(req, it);
 832                 rc = obd_get_request_slot(&obddev->u.cli);
 833                 if (rc != 0) {
 834                         mdc_put_mod_rpc_slot(req, it);
 835                         mdc_clear_replay_flag(req, 0);
 836                         ptlrpc_req_finished(req);
 837                         RETURN(rc);
 838                 }
 839         }
 840
 841         rc = ldlm_cli_enqueue(exp, &req, einfo, &res_id, policy, &flags, NULL,
 842                               0, lvb_type, lockh, 0);
 843         if (!it) {
 844                 /* For flock requests we immediatelly return without further
 845                    delay and let caller deal with the rest, since rest of
 846                    this function metadata processing makes no sense for flock
 847                    requests anyway. But in case of problem during comms with
 848                    Server (ETIMEDOUT) or any signal/kill attempt (EINTR), we
 849                    can not rely on caller and this mainly for F_UNLCKs
 850                    (explicits or automatically generated by Kernel to clean
 851                    current FLocks upon exit) that can't be trashed */
 852                 if (((rc == -EINTR) || (rc == -ETIMEDOUT)) &&
 853                     (einfo->ei_type == LDLM_FLOCK) &&
 854                     (einfo->ei_mode == LCK_NL))
 855                         goto resend;
 856                 RETURN(rc);
 857         }
 858
 859         obd_put_request_slot(&obddev->u.cli);
 860         mdc_put_mod_rpc_slot(req, it);
 861
 862         if (rc < 0) {
 863                 CDEBUG(D_INFO, "%s: ldlm_cli_enqueue failed: rc = %d\n",
 864                        obddev->obd_name, rc);
 865
 866                 mdc_clear_replay_flag(req, rc);
 867                 ptlrpc_req_finished(req);
 868                 RETURN(rc);
 869         }
 870
 871         lockrep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
 872         LASSERT(lockrep != NULL);
 873
 874         lockrep->lock_policy_res2 =
 875                 ptlrpc_status_ntoh(lockrep->lock_policy_res2);
 876
 877         /* Retry infinitely when the server returns -EINPROGRESS for the
 878          * intent operation, when server returns -EINPROGRESS for acquiring
 879          * intent lock, we'll retry in after_reply(). */
 880         if (it && (int)lockrep->lock_policy_res2 == -EINPROGRESS) {
 881                 mdc_clear_replay_flag(req, rc);
 882                 ptlrpc_req_finished(req);
 883                 resends++;
 884
 885                 CDEBUG(D_HA, "%s: resend:%d op:%d "DFID"/"DFID"\n",
 886                        obddev->obd_name, resends, it->it_op,
 887                        PFID(&op_data->op_fid1), PFID(&op_data->op_fid2));
 888
 889                 if (generation == obddev->u.cli.cl_import->imp_generation) {
 890                         goto resend;
 891                 } else {
 892                         CDEBUG(D_HA, "resend cross eviction\n");
 893                         RETURN(-EIO);
 894                 }
 895         }
 896
 897         rc = mdc_finish_enqueue(exp, req, einfo, it, lockh, rc);
 898         if (rc < 0) {
 899                 if (lustre_handle_is_used(lockh)) {
 900                         ldlm_lock_decref(lockh, einfo->ei_mode);
 901                         memset(lockh, 0, sizeof(*lockh));
 902                 }
 903                 ptlrpc_req_finished(req);
 904
 905                 it->d.lustre.it_lock_handle = 0;
 906                 it->d.lustre.it_lock_mode = 0;
 907                 it->d.lustre.it_data = NULL;
 908         }
 909
 910         RETURN(rc);
 911 }
 912
 913 static int mdc_finish_intent_lock(struct obd_export *exp,
 914                                   struct ptlrpc_request *request,
 915                                   struct md_op_data *op_data,
 916                                   struct lookup_intent *it,
 917                                   struct lustre_handle *lockh)
 918 {
 919         struct lustre_handle old_lock;
 920         struct mdt_body *mdt_body;
 921         struct ldlm_lock *lock;
 922         int rc;
 923         ENTRY;
 924
 925         LASSERT(request != NULL);
 926         LASSERT(request != LP_POISON);
 927         LASSERT(request->rq_repmsg != LP_POISON);
 928
 929         if (it->it_op & IT_READDIR)
 930                 RETURN(0);
 931
 932         if (!it_disposition(it, DISP_IT_EXECD)) {
 933                 /* The server failed before it even started executing the
 934                  * intent, i.e. because it couldn't unpack the request. */
 935                 LASSERT(it->d.lustre.it_status != 0);
 936                 RETURN(it->d.lustre.it_status);
 937         }
 938         rc = it_open_error(DISP_IT_EXECD, it);
 939         if (rc)
 940                 RETURN(rc);
 941
 942         mdt_body = req_capsule_server_get(&request->rq_pill, &RMF_MDT_BODY);
 943         LASSERT(mdt_body != NULL);      /* mdc_enqueue checked */
 944
 945         rc = it_open_error(DISP_LOOKUP_EXECD, it);
 946         if (rc)
 947                 RETURN(rc);
 948
 949         /* keep requests around for the multiple phases of the call
 950          * this shows the DISP_XX must guarantee we make it into the call
 951          */
 952         if (!it_disposition(it, DISP_ENQ_CREATE_REF) &&
 953             it_disposition(it, DISP_OPEN_CREATE) &&
 954             !it_open_error(DISP_OPEN_CREATE, it)) {
 955                 it_set_disposition(it, DISP_ENQ_CREATE_REF);
 956                 ptlrpc_request_addref(request); /* balanced in ll_create_node */
 957         }
 958         if (!it_disposition(it, DISP_ENQ_OPEN_REF) &&
 959             it_disposition(it, DISP_OPEN_OPEN) &&
 960             !it_open_error(DISP_OPEN_OPEN, it)) {
 961                 it_set_disposition(it, DISP_ENQ_OPEN_REF);
 962                 ptlrpc_request_addref(request); /* balanced in ll_file_open */
 963                 /* BUG 11546 - eviction in the middle of open rpc processing */
 964                 OBD_FAIL_TIMEOUT(OBD_FAIL_MDC_ENQUEUE_PAUSE, obd_timeout);
 965         }
 966
 967         if (it->it_op & IT_CREAT) {
 968                 /* XXX this belongs in ll_create_it */
 969         } else if (it->it_op == IT_OPEN) {
 970                 LASSERT(!it_disposition(it, DISP_OPEN_CREATE));
 971         } else {
 972                 LASSERT(it->it_op & (IT_GETATTR | IT_LOOKUP | IT_LAYOUT));
 973         }
 974
 975         /* If we already have a matching lock, then cancel the new
 976          * one.  We have to set the data here instead of in
 977          * mdc_enqueue, because we need to use the child's inode as
 978          * the l_ast_data to match, and that's not available until
 979          * intent_finish has performed the iget().) */
 980         lock = ldlm_handle2lock(lockh);
 981         if (lock) {
 982                 ldlm_policy_data_t policy = lock->l_policy_data;
 983                 LDLM_DEBUG(lock, "matching against this");
 984
 985                 LASSERTF(fid_res_name_eq(&mdt_body->mbo_fid1,
 986                                          &lock->l_resource->lr_name),
 987                          "Lock res_id: "DLDLMRES", fid: "DFID"\n",
 988                          PLDLMRES(lock->l_resource), PFID(&mdt_body->mbo_fid1));
 989                 LDLM_LOCK_PUT(lock);
 990
 991                 memcpy(&old_lock, lockh, sizeof(*lockh));
 992                 if (ldlm_lock_match(NULL, LDLM_FL_BLOCK_GRANTED, NULL,
 993                                     LDLM_IBITS, &policy, LCK_NL, &old_lock, 0)) {
 994                         ldlm_lock_decref_and_cancel(lockh,
 995                                                     it->d.lustre.it_lock_mode);
 996                         memcpy(lockh, &old_lock, sizeof(old_lock));
 997                         it->d.lustre.it_lock_handle = lockh->cookie;
 998                 }
 999         }
1000         CDEBUG(D_DENTRY,"D_IT dentry %.*s intent: %s status %d disp %x rc %d\n",
1001                 (int)op_data->op_namelen, op_data->op_name,
1002                 ldlm_it2str(it->it_op), it->d.lustre.it_status,
1003                 it->d.lustre.it_disposition, rc);
1004         RETURN(rc);
1005 }
1006
1007 int mdc_revalidate_lock(struct obd_export *exp, struct lookup_intent *it,
1008                         struct lu_fid *fid, __u64 *bits)
1009 {
1010         /* We could just return 1 immediately, but since we should only
1011          * be called in revalidate_it if we already have a lock, let's
1012          * verify that. */
1013         struct ldlm_res_id res_id;
1014         struct lustre_handle lockh;
1015         ldlm_policy_data_t policy;
1016         ldlm_mode_t mode;
1017         ENTRY;
1018
1019         if (it->d.lustre.it_lock_handle) {
1020                 lockh.cookie = it->d.lustre.it_lock_handle;
1021                 mode = ldlm_revalidate_lock_handle(&lockh, bits);
1022         } else {
1023                 fid_build_reg_res_name(fid, &res_id);
1024                 switch (it->it_op) {
1025                 case IT_GETATTR:
1026                         /* File attributes are held under multiple bits:
1027                          * nlink is under lookup lock, size and times are
1028                          * under UPDATE lock and recently we've also got
1029                          * a separate permissions lock for owner/group/acl that
1030                          * were protected by lookup lock before.
1031                          * Getattr must provide all of that information,
1032                          * so we need to ensure we have all of those locks.
1033                          * Unfortunately, if the bits are split across multiple
1034                          * locks, there's no easy way to match all of them here,
1035                          * so an extra RPC would be performed to fetch all
1036                          * of those bits at once for now. */
1037                         /* For new MDTs(> 2.4), UPDATE|PERM should be enough,
1038                          * but for old MDTs (< 2.4), permission is covered
1039                          * by LOOKUP lock, so it needs to match all bits here.*/
1040                         policy.l_inodebits.bits = MDS_INODELOCK_UPDATE |
1041                                                   MDS_INODELOCK_LOOKUP |
1042                                                   MDS_INODELOCK_PERM;
1043                         break;
1044                 case IT_READDIR:
1045                         policy.l_inodebits.bits = MDS_INODELOCK_UPDATE;
1046                         break;
1047                 case IT_LAYOUT:
1048                         policy.l_inodebits.bits = MDS_INODELOCK_LAYOUT;
1049                         break;
1050                 default:
1051                         policy.l_inodebits.bits = MDS_INODELOCK_LOOKUP;
1052                         break;
1053                 }
1054
1055                 mode = mdc_lock_match(exp, LDLM_FL_BLOCK_GRANTED, fid,
1056                                       LDLM_IBITS, &policy,
1057                                       LCK_CR | LCK_CW | LCK_PR | LCK_PW,
1058                                       &lockh);
1059         }
1060
1061         if (mode) {
1062                 it->d.lustre.it_lock_handle = lockh.cookie;
1063                 it->d.lustre.it_lock_mode = mode;
1064         } else {
1065                 it->d.lustre.it_lock_handle = 0;
1066                 it->d.lustre.it_lock_mode = 0;
1067         }
1068
1069         RETURN(!!mode);
1070 }
1071
1072 /*
1073  * This long block is all about fixing up the lock and request state
1074  * so that it is correct as of the moment _before_ the operation was
1075  * applied; that way, the VFS will think that everything is normal and
1076  * call Lustre's regular VFS methods.
1077  *
1078  * If we're performing a creation, that means that unless the creation
1079  * failed with EEXIST, we should fake up a negative dentry.
1080  *
1081  * For everything else, we want to lookup to succeed.
1082  *
1083  * One additional note: if CREATE or OPEN succeeded, we add an extra
1084  * reference to the request because we need to keep it around until
1085  * ll_create/ll_open gets called.
1086  *
1087  * The server will return to us, in it_disposition, an indication of
1088  * exactly what d.lustre.it_status refers to.
1089  *
1090  * If DISP_OPEN_OPEN is set, then d.lustre.it_status refers to the open() call,
1091  * otherwise if DISP_OPEN_CREATE is set, then it status is the
1092  * creation failure mode.  In either case, one of DISP_LOOKUP_NEG or
1093  * DISP_LOOKUP_POS will be set, indicating whether the child lookup
1094  * was successful.
1095  *
1096  * Else, if DISP_LOOKUP_EXECD then d.lustre.it_status is the rc of the
1097  * child lookup.
1098  */
1099 int mdc_intent_lock(struct obd_export *exp, struct md_op_data *op_data,
1100                     struct lookup_intent *it, struct ptlrpc_request **reqp,
1101                     ldlm_blocking_callback cb_blocking, __u64 extra_lock_flags)
1102 {
1103         struct ldlm_enqueue_info einfo = {
1104                 .ei_type        = LDLM_IBITS,
1105                 .ei_mode        = it_to_lock_mode(it),
1106                 .ei_cb_bl       = cb_blocking,
1107                 .ei_cb_cp       = ldlm_completion_ast,
1108         };
1109         struct lustre_handle lockh;
1110         int rc = 0;
1111         ENTRY;
1112         LASSERT(it);
1113
1114         CDEBUG(D_DLMTRACE, "(name: %.*s,"DFID") in obj "DFID
1115                 ", intent: %s flags %#"LPF64"o\n", (int)op_data->op_namelen,
1116                 op_data->op_name, PFID(&op_data->op_fid2),
1117                 PFID(&op_data->op_fid1), ldlm_it2str(it->it_op),
1118                 it->it_flags);
1119
1120         lockh.cookie = 0;
1121         if (fid_is_sane(&op_data->op_fid2) &&
1122             (it->it_op & (IT_LOOKUP | IT_GETATTR | IT_READDIR))) {
1123                 /* We could just return 1 immediately, but since we should only
1124                  * be called in revalidate_it if we already have a lock, let's
1125                  * verify that. */
1126                 it->d.lustre.it_lock_handle = 0;
1127                 rc = mdc_revalidate_lock(exp, it, &op_data->op_fid2, NULL);
1128                 /* Only return failure if it was not GETATTR by cfid
1129                    (from inode_revalidate) */
1130                 if (rc || op_data->op_namelen != 0)
1131                         RETURN(rc);
1132         }
1133
1134         /* For case if upper layer did not alloc fid, do it now. */
1135         if (!fid_is_sane(&op_data->op_fid2) && it->it_op & IT_CREAT) {
1136                 rc = mdc_fid_alloc(NULL, exp, &op_data->op_fid2, op_data);
1137                 if (rc < 0) {
1138                         CERROR("Can't alloc new fid, rc %d\n", rc);
1139                         RETURN(rc);
1140                 }
1141         }
1142
1143         rc = mdc_enqueue(exp, &einfo, NULL, it, op_data, &lockh,
1144                          extra_lock_flags);
1145         if (rc < 0)
1146                 RETURN(rc);
1147
1148         *reqp = it->d.lustre.it_data;
1149         rc = mdc_finish_intent_lock(exp, *reqp, op_data, it, &lockh);
1150         RETURN(rc);
1151 }
1152
1153 static int mdc_intent_getattr_async_interpret(const struct lu_env *env,
1154                                               struct ptlrpc_request *req,
1155                                               void *args, int rc)
1156 {
1157         struct mdc_getattr_args  *ga = args;
1158         struct obd_export        *exp = ga->ga_exp;
1159         struct md_enqueue_info   *minfo = ga->ga_minfo;
1160         struct ldlm_enqueue_info *einfo = ga->ga_einfo;
1161         struct lookup_intent     *it;
1162         struct lustre_handle     *lockh;
1163         struct obd_device        *obddev;
1164         struct ldlm_reply        *lockrep;
1165         __u64                     flags = LDLM_FL_HAS_INTENT;
1166         ENTRY;
1167
1168         it    = &minfo->mi_it;
1169         lockh = &minfo->mi_lockh;
1170
1171         obddev = class_exp2obd(exp);
1172
1173         obd_put_request_slot(&obddev->u.cli);
1174         if (OBD_FAIL_CHECK(OBD_FAIL_MDC_GETATTR_ENQUEUE))
1175                 rc = -ETIMEDOUT;
1176
1177         rc = ldlm_cli_enqueue_fini(exp, req, einfo->ei_type, 1, einfo->ei_mode,
1178                                    &flags, NULL, 0, lockh, rc);
1179         if (rc < 0) {
1180                 CERROR("ldlm_cli_enqueue_fini: %d\n", rc);
1181                 mdc_clear_replay_flag(req, rc);
1182                 GOTO(out, rc);
1183         }
1184
1185         lockrep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
1186         LASSERT(lockrep != NULL);
1187
1188         lockrep->lock_policy_res2 =
1189                 ptlrpc_status_ntoh(lockrep->lock_policy_res2);
1190
1191         rc = mdc_finish_enqueue(exp, req, einfo, it, lockh, rc);
1192         if (rc)
1193                 GOTO(out, rc);
1194
1195         rc = mdc_finish_intent_lock(exp, req, &minfo->mi_data, it, lockh);
1196         EXIT;
1197
1198 out:
1199         OBD_FREE_PTR(einfo);
1200         minfo->mi_cb(req, minfo, rc);
1201         return 0;
1202 }
1203
1204 int mdc_intent_getattr_async(struct obd_export *exp,
1205                              struct md_enqueue_info *minfo,
1206                              struct ldlm_enqueue_info *einfo)
1207 {
1208         struct md_op_data       *op_data = &minfo->mi_data;
1209         struct lookup_intent    *it = &minfo->mi_it;
1210         struct ptlrpc_request   *req;
1211         struct mdc_getattr_args *ga;
1212         struct obd_device       *obddev = class_exp2obd(exp);
1213         struct ldlm_res_id       res_id;
1214         /*XXX: Both MDS_INODELOCK_LOOKUP and MDS_INODELOCK_UPDATE are needed
1215          *     for statahead currently. Consider CMD in future, such two bits
1216          *     maybe managed by different MDS, should be adjusted then. */
1217         ldlm_policy_data_t       policy = {
1218                                         .l_inodebits = { MDS_INODELOCK_LOOKUP |
1219                                                          MDS_INODELOCK_UPDATE }
1220                                  };
1221         int                      rc = 0;
1222         __u64                    flags = LDLM_FL_HAS_INTENT;
1223         ENTRY;
1224
1225         CDEBUG(D_DLMTRACE, "name: %.*s in inode "DFID", intent: %s flags %#"
1226                 LPF64"o\n",
1227                 (int)op_data->op_namelen, op_data->op_name,
1228                 PFID(&op_data->op_fid1), ldlm_it2str(it->it_op), it->it_flags);
1229
1230         fid_build_reg_res_name(&op_data->op_fid1, &res_id);
1231         req = mdc_intent_getattr_pack(exp, it, op_data);
1232         if (IS_ERR(req))
1233                 RETURN(PTR_ERR(req));
1234
1235         rc = obd_get_request_slot(&obddev->u.cli);
1236         if (rc != 0) {
1237                 ptlrpc_req_finished(req);
1238                 RETURN(rc);
1239         }
1240
1241         rc = ldlm_cli_enqueue(exp, &req, einfo, &res_id, &policy, &flags, NULL,
1242                               0, LVB_T_NONE, &minfo->mi_lockh, 1);
1243         if (rc < 0) {
1244                 obd_put_request_slot(&obddev->u.cli);
1245                 ptlrpc_req_finished(req);
1246                 RETURN(rc);
1247         }
1248
1249         CLASSERT(sizeof(*ga) <= sizeof(req->rq_async_args));
1250         ga = ptlrpc_req_async_args(req);
1251         ga->ga_exp = exp;
1252         ga->ga_minfo = minfo;
1253         ga->ga_einfo = einfo;
1254
1255         req->rq_interpret_reply = mdc_intent_getattr_async_interpret;
1256         ptlrpcd_add_req(req);
1257
1258         RETURN(0);
1259 }