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