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.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  */
  31
  32 #define DEBUG_SUBSYSTEM S_MDC
  33
  34 #include <linux/module.h>
  35
  36 #include <obd.h>
  37 #include <obd_class.h>
  38 #include <lustre_dlm.h>
  39 #include <lustre_fid.h>
  40 #include <lustre_intent.h>
  41 #include <lustre_mdc.h>
  42 #include <lustre_net.h>
  43 #include <lustre_req_layout.h>
  44 #include <lustre_swab.h>
  45 #include <lustre_acl.h>
  46
  47 #include "mdc_internal.h"
  48
  49 struct mdc_getattr_args {
  50         struct obd_export       *ga_exp;
  51         struct md_op_item       *ga_item;
  52 };
  53
  54 int it_open_error(int phase, struct lookup_intent *it)
  55 {
  56         if (it_disposition(it, DISP_OPEN_LEASE)) {
  57                 if (phase >= DISP_OPEN_LEASE)
  58                         return it->it_status;
  59                 else
  60                         return 0;
  61         }
  62         if (it_disposition(it, DISP_OPEN_OPEN)) {
  63                 if (phase >= DISP_OPEN_OPEN)
  64                         return it->it_status;
  65                 else
  66                         return 0;
  67         }
  68
  69         if (it_disposition(it, DISP_OPEN_CREATE)) {
  70                 if (phase >= DISP_OPEN_CREATE)
  71                         return it->it_status;
  72                 else
  73                         return 0;
  74         }
  75
  76         if (it_disposition(it, DISP_LOOKUP_EXECD)) {
  77                 if (phase >= DISP_LOOKUP_EXECD)
  78                         return it->it_status;
  79                 else
  80                         return 0;
  81         }
  82
  83         if (it_disposition(it, DISP_IT_EXECD)) {
  84                 if (phase >= DISP_IT_EXECD)
  85                         return it->it_status;
  86                 else
  87                         return 0;
  88         }
  89
  90         CERROR("it disp: %X, status: %d\n", it->it_disposition, it->it_status);
  91         LBUG();
  92
  93         return 0;
  94 }
  95 EXPORT_SYMBOL(it_open_error);
  96
  97 /* this must be called on a lockh that is known to have a referenced lock */
  98 int mdc_set_lock_data(struct obd_export *exp, const struct lustre_handle *lockh,
  99                       void *data, __u64 *bits)
 100 {
 101         struct ldlm_lock *lock;
 102         struct inode *new_inode = data;
 103
 104         ENTRY;
 105         if (bits)
 106                 *bits = 0;
 107
 108         if (!lustre_handle_is_used(lockh))
 109                 RETURN(0);
 110
 111         lock = ldlm_handle2lock(lockh);
 112
 113         LASSERT(lock != NULL);
 114         lock_res_and_lock(lock);
 115         if (lock->l_resource->lr_lvb_inode &&
 116             lock->l_resource->lr_lvb_inode != data) {
 117                 struct inode *old_inode = lock->l_resource->lr_lvb_inode;
 118
 119                 LASSERTF(old_inode->i_state & I_FREEING,
 120                          "Found existing inode %p/%lu/%u state %lu in lock: setting data to %p/%lu/%u\n",
 121                          old_inode, old_inode->i_ino, old_inode->i_generation,
 122                          old_inode->i_state,
 123                          new_inode, new_inode->i_ino, new_inode->i_generation);
 124         }
 125         lock->l_resource->lr_lvb_inode = new_inode;
 126         if (bits)
 127                 *bits = lock->l_policy_data.l_inodebits.bits;
 128
 129         unlock_res_and_lock(lock);
 130         LDLM_LOCK_PUT(lock);
 131
 132         RETURN(0);
 133 }
 134
 135 enum ldlm_mode mdc_lock_match(struct obd_export *exp, __u64 flags,
 136                               const struct lu_fid *fid, enum ldlm_type type,
 137                               union ldlm_policy_data *policy,
 138                               enum ldlm_mode mode, struct lustre_handle *lockh)
 139 {
 140         struct ldlm_res_id res_id;
 141         enum ldlm_mode rc;
 142
 143         ENTRY;
 144         fid_build_reg_res_name(fid, &res_id);
 145         /* LU-4405: Clear bits not supported by server */
 146         policy->l_inodebits.bits &= exp_connect_ibits(exp);
 147         rc = ldlm_lock_match(class_exp2obd(exp)->obd_namespace, flags,
 148                              &res_id, type, policy, mode, lockh);
 149         RETURN(rc);
 150 }
 151
 152 int mdc_cancel_unused(struct obd_export *exp, const struct lu_fid *fid,
 153                       union ldlm_policy_data *policy, enum ldlm_mode mode,
 154                       enum ldlm_cancel_flags flags, void *opaque)
 155 {
 156         struct obd_device *obd = class_exp2obd(exp);
 157         struct ldlm_res_id res_id;
 158         int rc;
 159
 160         ENTRY;
 161         fid_build_reg_res_name(fid, &res_id);
 162         rc = ldlm_cli_cancel_unused_resource(obd->obd_namespace, &res_id,
 163                                              policy, mode, flags, opaque);
 164         RETURN(rc);
 165 }
 166
 167 int mdc_null_inode(struct obd_export *exp,
 168                    const struct lu_fid *fid)
 169 {
 170         struct ldlm_res_id res_id;
 171         struct ldlm_resource *res;
 172         struct ldlm_namespace *ns = class_exp2obd(exp)->obd_namespace;
 173
 174         ENTRY;
 175         LASSERTF(ns != NULL, "no namespace passed\n");
 176
 177         fid_build_reg_res_name(fid, &res_id);
 178
 179         res = ldlm_resource_get(ns, &res_id, 0, 0);
 180         if (IS_ERR(res))
 181                 RETURN(0);
 182
 183         lock_res(res);
 184         res->lr_lvb_inode = NULL;
 185         unlock_res(res);
 186
 187         ldlm_resource_putref(res);
 188         RETURN(0);
 189 }
 190
 191 static inline void mdc_clear_replay_flag(struct ptlrpc_request *req, int rc)
 192 {
 193         /* Don't hold error requests for replay. */
 194         if (req->rq_replay) {
 195                 spin_lock(&req->rq_lock);
 196                 req->rq_replay = 0;
 197                 spin_unlock(&req->rq_lock);
 198         }
 199         if (rc && req->rq_transno != 0) {
 200                 DEBUG_REQ(D_ERROR, req, "transno returned on error: rc = %d",
 201                           rc);
 202                 LBUG();
 203         }
 204 }
 205
 206 /**
 207  * Save a large LOV EA into the request buffer so that it is available
 208  * for replay.  We don't do this in the initial request because the
 209  * original request doesn't need this buffer (at most it sends just the
 210  * lov_mds_md) and it is a waste of RAM/bandwidth to send the empty
 211  * buffer and may also be difficult to allocate and save a very large
 212  * request buffer for each open. (b=5707)
 213  *
 214  * OOM here may cause recovery failure if lmm is needed (only for the
 215  * original open if the MDS crashed just when this client also OOM'd)
 216  * but this is incredibly unlikely, and questionable whether the client
 217  * could do MDS recovery under OOM anyways...
 218  */
 219 static int mdc_save_lovea(struct ptlrpc_request *req, void *data, u32 size)
 220 {
 221         struct req_capsule *pill = &req->rq_pill;
 222         void *lovea;
 223         int rc = 0;
 224
 225         if (req_capsule_get_size(pill, &RMF_EADATA, RCL_CLIENT) < size) {
 226                 rc = sptlrpc_cli_enlarge_reqbuf(req, &RMF_EADATA, size);
 227                 if (rc) {
 228                         CERROR("%s: Can't enlarge ea size to %d: rc = %d\n",
 229                                req->rq_export->exp_obd->obd_name,
 230                                size, rc);
 231                         return rc;
 232                 }
 233         } else {
 234                 req_capsule_shrink(pill, &RMF_EADATA, size, RCL_CLIENT);
 235         }
 236
 237         req_capsule_set_size(pill, &RMF_EADATA, RCL_CLIENT, size);
 238         lovea = req_capsule_client_get(pill, &RMF_EADATA);
 239         if (lovea) {
 240                 memcpy(lovea, data, size);
 241                 lov_fix_ea_for_replay(lovea);
 242         }
 243
 244         return rc;
 245 }
 246
 247 static struct ptlrpc_request *
 248 mdc_intent_open_pack(struct obd_export *exp, struct lookup_intent *it,
 249                      struct md_op_data *op_data, __u32 acl_bufsize)
 250 {
 251         struct ptlrpc_request *req;
 252         struct obd_device *obd = class_exp2obd(exp);
 253         struct ldlm_intent *lit;
 254         const void *lmm = op_data->op_data;
 255         __u32 lmmsize = op_data->op_data_size;
 256         __u32  mdt_md_capsule_size;
 257         LIST_HEAD(cancels);
 258         int count = 0;
 259         enum ldlm_mode mode;
 260         int repsize, repsize_estimate;
 261         int rc;
 262
 263         ENTRY;
 264
 265         mdt_md_capsule_size = obd->u.cli.cl_default_mds_easize;
 266
 267         it->it_create_mode = (it->it_create_mode & ~S_IFMT) | S_IFREG;
 268
 269         /* XXX: openlock is not cancelled for cross-refs. */
 270         /* If inode is known, cancel conflicting OPEN locks. */
 271         if (fid_is_sane(&op_data->op_fid2)) {
 272                 if (it->it_flags & MDS_OPEN_LEASE) { /* try to get lease */
 273                         if (it->it_flags & MDS_FMODE_WRITE)
 274                                 mode = LCK_EX;
 275                         else
 276                                 mode = LCK_PR;
 277                 } else {
 278                         if (it->it_flags & (MDS_FMODE_WRITE | MDS_OPEN_TRUNC))
 279                                 mode = LCK_CW;
 280 #ifdef FMODE_EXEC
 281                         else if (it->it_flags & FMODE_EXEC)
 282                                 mode = LCK_PR;
 283 #endif
 284                         else
 285                                 mode = LCK_CR;
 286                 }
 287                 count = mdc_resource_get_unused(exp, &op_data->op_fid2,
 288                                                 &cancels, mode,
 289                                                 MDS_INODELOCK_OPEN);
 290         }
 291
 292         /* If CREATE, cancel parent's UPDATE lock. */
 293         if (it->it_op & IT_CREAT)
 294                 mode = LCK_EX;
 295         else
 296                 mode = LCK_CR;
 297         count += mdc_resource_get_unused(exp, &op_data->op_fid1,
 298                                          &cancels, mode,
 299                                          MDS_INODELOCK_UPDATE);
 300
 301         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 302                                    &RQF_LDLM_INTENT_OPEN);
 303         if (req == NULL) {
 304                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
 305                 RETURN(ERR_PTR(-ENOMEM));
 306         }
 307
 308         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
 309                              op_data->op_namelen + 1);
 310         if (cl_is_lov_delay_create(it->it_flags)) {
 311                 /* open(O_LOV_DELAY_CREATE) won't pack lmm */
 312                 LASSERT(lmmsize == 0);
 313                 req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT, 0);
 314         } else {
 315                 req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT,
 316                              max(lmmsize, obd->u.cli.cl_default_mds_easize));
 317         }
 318
 319         req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX_NAME,
 320                              RCL_CLIENT, op_data->op_file_secctx_name != NULL ?
 321                              op_data->op_file_secctx_name_size : 0);
 322
 323         req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX, RCL_CLIENT,
 324                              op_data->op_file_secctx_size);
 325
 326         req_capsule_set_size(&req->rq_pill, &RMF_FILE_ENCCTX, RCL_CLIENT,
 327                              op_data->op_file_encctx_size);
 328
 329         /* get SELinux policy info if any */
 330         rc = sptlrpc_get_sepol(req);
 331         if (rc < 0) {
 332                 ptlrpc_request_free(req);
 333                 RETURN(ERR_PTR(rc));
 334         }
 335         req_capsule_set_size(&req->rq_pill, &RMF_SELINUX_POL, RCL_CLIENT,
 336                              strlen(req->rq_sepol) ?
 337                              strlen(req->rq_sepol) + 1 : 0);
 338
 339         rc = ldlm_prep_enqueue_req(exp, req, &cancels, count);
 340         if (rc < 0) {
 341                 ptlrpc_request_free(req);
 342                 RETURN(ERR_PTR(rc));
 343         }
 344
 345         spin_lock(&req->rq_lock);
 346         req->rq_replay = req->rq_import->imp_replayable;
 347         spin_unlock(&req->rq_lock);
 348
 349         /* pack the intent */
 350         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
 351         lit->opc = (__u64)it->it_op;
 352
 353         /* pack the intended request */
 354         mdc_open_pack(&req->rq_pill, op_data, it->it_create_mode, 0,
 355                       it->it_flags, lmm, lmmsize);
 356
 357         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
 358                              mdt_md_capsule_size);
 359         req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER, acl_bufsize);
 360
 361         if (!(it->it_op & IT_CREAT) && it->it_op & IT_OPEN &&
 362             req_capsule_has_field(&req->rq_pill, &RMF_FILE_SECCTX_NAME,
 363                                   RCL_CLIENT) &&
 364             op_data->op_file_secctx_name_size > 0 &&
 365             op_data->op_file_secctx_name != NULL) {
 366                 char *secctx_name;
 367
 368                 secctx_name = req_capsule_client_get(&req->rq_pill,
 369                                                      &RMF_FILE_SECCTX_NAME);
 370                 memcpy(secctx_name, op_data->op_file_secctx_name,
 371                        op_data->op_file_secctx_name_size);
 372                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX,
 373                                      RCL_SERVER,
 374                                      obd->u.cli.cl_max_mds_easize);
 375
 376                 CDEBUG(D_SEC, "packed '%.*s' as security xattr name\n",
 377                        op_data->op_file_secctx_name_size,
 378                        op_data->op_file_secctx_name);
 379
 380         } else {
 381                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX,
 382                                      RCL_SERVER, 0);
 383         }
 384
 385         if (exp_connect_encrypt(exp) && !(it->it_op & IT_CREAT) &&
 386             it->it_op & IT_OPEN)
 387                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_ENCCTX,
 388                                      RCL_SERVER,
 389                                      obd->u.cli.cl_max_mds_easize);
 390         else
 391                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_ENCCTX,
 392                                      RCL_SERVER, 0);
 393
 394         /**
 395          * Inline buffer for possible data from Data-on-MDT files.
 396          */
 397         req_capsule_set_size(&req->rq_pill, &RMF_NIOBUF_INLINE, RCL_SERVER,
 398                              sizeof(struct niobuf_remote));
 399         req_capsule_set_size(&req->rq_pill, &RMF_DEFAULT_MDT_MD, RCL_SERVER,
 400                              sizeof(struct lmv_user_md));
 401         ptlrpc_request_set_replen(req);
 402
 403         /* Get real repbuf allocated size as rounded up power of 2 */
 404         repsize = size_roundup_power2(req->rq_replen +
 405                                       lustre_msg_early_size);
 406         /* Estimate free space for DoM files in repbuf */
 407         repsize_estimate = repsize - (req->rq_replen -
 408                            mdt_md_capsule_size +
 409                            sizeof(struct lov_comp_md_v1) +
 410                            sizeof(struct lov_comp_md_entry_v1) +
 411                            lov_mds_md_size(0, LOV_MAGIC_V3));
 412
 413         if (repsize_estimate < obd->u.cli.cl_dom_min_inline_repsize) {
 414                 repsize = obd->u.cli.cl_dom_min_inline_repsize -
 415                           repsize_estimate + sizeof(struct niobuf_remote);
 416                 req_capsule_set_size(&req->rq_pill, &RMF_NIOBUF_INLINE,
 417                                      RCL_SERVER,
 418                                      sizeof(struct niobuf_remote) + repsize);
 419                 ptlrpc_request_set_replen(req);
 420                 CDEBUG(D_INFO, "Increase repbuf by %d bytes, total: %d\n",
 421                        repsize, req->rq_replen);
 422                 repsize = size_roundup_power2(req->rq_replen +
 423                                               lustre_msg_early_size);
 424         }
 425         /* The only way to report real allocated repbuf size to the server
 426          * is the lm_repsize but it must be set prior buffer allocation itself
 427          * due to security reasons - it is part of buffer used in signature
 428          * calculation (see LU-11414). Therefore the saved size is predicted
 429          * value as rq_replen rounded to the next higher power of 2.
 430          * Such estimation is safe. Though the final allocated buffer might
 431          * be even larger, it is not possible to know that at this point.
 432          */
 433         req->rq_reqmsg->lm_repsize = repsize;
 434         RETURN(req);
 435 }
 436
 437 #define GA_DEFAULT_EA_NAME_LEN   20
 438 #define GA_DEFAULT_EA_VAL_LEN   250
 439 #define GA_DEFAULT_EA_NUM        10
 440
 441 static struct ptlrpc_request *
 442 mdc_intent_getxattr_pack(struct obd_export *exp, struct lookup_intent *it,
 443                          struct md_op_data *op_data)
 444 {
 445         struct ptlrpc_request *req;
 446         struct ldlm_intent *lit;
 447         int rc, count = 0;
 448         LIST_HEAD(cancels);
 449         u32 ea_vals_buf_size = GA_DEFAULT_EA_VAL_LEN * GA_DEFAULT_EA_NUM;
 450
 451         ENTRY;
 452         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 453                                         &RQF_LDLM_INTENT_GETXATTR);
 454         if (req == NULL)
 455                 RETURN(ERR_PTR(-ENOMEM));
 456
 457         /* get SELinux policy info if any */
 458         rc = sptlrpc_get_sepol(req);
 459         if (rc < 0) {
 460                 ptlrpc_request_free(req);
 461                 RETURN(ERR_PTR(rc));
 462         }
 463         req_capsule_set_size(&req->rq_pill, &RMF_SELINUX_POL, RCL_CLIENT,
 464                              strlen(req->rq_sepol) ?
 465                              strlen(req->rq_sepol) + 1 : 0);
 466
 467         rc = ldlm_prep_enqueue_req(exp, req, &cancels, count);
 468         if (rc) {
 469                 ptlrpc_request_free(req);
 470                 RETURN(ERR_PTR(rc));
 471         }
 472
 473         /* pack the intent */
 474         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
 475         lit->opc = IT_GETXATTR;
 476         /* Message below is checked in sanity-selinux test_20d
 477          * and sanity-sec test_49
 478          */
 479         CDEBUG(D_INFO, "%s: get xattrs for "DFID"\n",
 480                exp->exp_obd->obd_name, PFID(&op_data->op_fid1));
 481
 482 #if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(3, 0, 53, 0)
 483         /* If the supplied buffer is too small then the server will return
 484          * -ERANGE and llite will fallback to using non cached xattr
 485          * operations. On servers before 2.10.1 a (non-cached) listxattr RPC
 486          * for an orphan or dead file causes an oops. So let's try to avoid
 487          * sending too small a buffer to too old a server. This is effectively
 488          * undoing the memory conservation of LU-9417 when it would be *more*
 489          * likely to crash the server. See LU-9856.
 490          */
 491         if (exp->exp_connect_data.ocd_version < OBD_OCD_VERSION(2, 10, 1, 0))
 492                 ea_vals_buf_size = max_t(u32, ea_vals_buf_size,
 493                                          exp->exp_connect_data.ocd_max_easize);
 494 #endif
 495
 496         /* pack the intended request */
 497         mdc_pack_body(&req->rq_pill, &op_data->op_fid1, op_data->op_valid,
 498                       ea_vals_buf_size, -1, 0);
 499
 500         /* get SELinux policy info if any */
 501         mdc_file_sepol_pack(&req->rq_pill);
 502
 503         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_SERVER,
 504                              GA_DEFAULT_EA_NAME_LEN * GA_DEFAULT_EA_NUM);
 505
 506         req_capsule_set_size(&req->rq_pill, &RMF_EAVALS, RCL_SERVER,
 507                              ea_vals_buf_size);
 508
 509         req_capsule_set_size(&req->rq_pill, &RMF_EAVALS_LENS, RCL_SERVER,
 510                              sizeof(u32) * GA_DEFAULT_EA_NUM);
 511
 512         req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER, 0);
 513
 514         ptlrpc_request_set_replen(req);
 515
 516         RETURN(req);
 517 }
 518
 519 static struct ptlrpc_request *
 520 mdc_intent_getattr_pack(struct obd_export *exp, struct lookup_intent *it,
 521                         struct md_op_data *op_data, __u32 acl_bufsize)
 522 {
 523         struct ptlrpc_request *req;
 524         struct obd_device *obd = class_exp2obd(exp);
 525         u64 valid = OBD_MD_FLGETATTR | OBD_MD_FLEASIZE | OBD_MD_FLMODEASIZE |
 526                     OBD_MD_FLDIREA | OBD_MD_MEA | OBD_MD_FLACL |
 527                     OBD_MD_DEFAULT_MEA;
 528         struct ldlm_intent *lit;
 529         __u32 easize;
 530         bool have_secctx = false;
 531         int rc;
 532
 533         ENTRY;
 534         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 535                                    &RQF_LDLM_INTENT_GETATTR);
 536         if (req == NULL)
 537                 RETURN(ERR_PTR(-ENOMEM));
 538
 539         /* send name of security xattr to get upon intent */
 540         if (it->it_op & (IT_LOOKUP | IT_GETATTR) &&
 541             req_capsule_has_field(&req->rq_pill, &RMF_FILE_SECCTX_NAME,
 542                                   RCL_CLIENT) &&
 543             op_data->op_file_secctx_name_size > 0 &&
 544             op_data->op_file_secctx_name != NULL) {
 545                 have_secctx = true;
 546                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX_NAME,
 547                                      RCL_CLIENT,
 548                                      op_data->op_file_secctx_name_size);
 549         }
 550
 551         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
 552                              op_data->op_namelen + 1);
 553
 554         rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
 555         if (rc) {
 556                 ptlrpc_request_free(req);
 557                 RETURN(ERR_PTR(rc));
 558         }
 559
 560         /* pack the intent */
 561         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
 562         lit->opc = (__u64)it->it_op;
 563
 564         if (obd->u.cli.cl_default_mds_easize > 0)
 565                 easize = obd->u.cli.cl_default_mds_easize;
 566         else
 567                 easize = obd->u.cli.cl_max_mds_easize;
 568
 569         /* pack the intended request */
 570         mdc_getattr_pack(&req->rq_pill, valid, it->it_flags, op_data, easize);
 571
 572         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER, easize);
 573         req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER, acl_bufsize);
 574         req_capsule_set_size(&req->rq_pill, &RMF_DEFAULT_MDT_MD, RCL_SERVER,
 575                              sizeof(struct lmv_user_md));
 576
 577         if (have_secctx) {
 578                 char *secctx_name;
 579
 580                 secctx_name = req_capsule_client_get(&req->rq_pill,
 581                                                      &RMF_FILE_SECCTX_NAME);
 582                 memcpy(secctx_name, op_data->op_file_secctx_name,
 583                        op_data->op_file_secctx_name_size);
 584
 585                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX,
 586                                      RCL_SERVER, easize);
 587
 588                 CDEBUG(D_SEC, "packed '%.*s' as security xattr name\n",
 589                        op_data->op_file_secctx_name_size,
 590                        op_data->op_file_secctx_name);
 591         } else {
 592                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX,
 593                                      RCL_SERVER, 0);
 594         }
 595
 596         if (exp_connect_encrypt(exp) && it->it_op & (IT_LOOKUP | IT_GETATTR))
 597                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_ENCCTX,
 598                                      RCL_SERVER, easize);
 599         else
 600                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_ENCCTX,
 601                                      RCL_SERVER, 0);
 602
 603         ptlrpc_request_set_replen(req);
 604         RETURN(req);
 605 }
 606
 607 static struct ptlrpc_request *mdc_intent_layout_pack(struct obd_export *exp,
 608                                                      struct lookup_intent *it,
 609                                                      struct md_op_data *op_data)
 610 {
 611         struct obd_device *obd = class_exp2obd(exp);
 612         struct ptlrpc_request *req;
 613         struct ldlm_intent *lit;
 614         struct layout_intent *layout;
 615         LIST_HEAD(cancels);
 616         int count = 0, rc;
 617
 618         ENTRY;
 619         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 620                                 &RQF_LDLM_INTENT_LAYOUT);
 621         if (req == NULL)
 622                 RETURN(ERR_PTR(-ENOMEM));
 623
 624         if (fid_is_sane(&op_data->op_fid2) && (it->it_op & IT_LAYOUT) &&
 625             (it->it_flags & FMODE_WRITE)) {
 626                 count = mdc_resource_get_unused(exp, &op_data->op_fid2,
 627                                                 &cancels, LCK_EX,
 628                                                 MDS_INODELOCK_LAYOUT);
 629         }
 630
 631         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT, 0);
 632         rc = ldlm_prep_enqueue_req(exp, req, &cancels, count);
 633         if (rc) {
 634                 ptlrpc_request_free(req);
 635                 RETURN(ERR_PTR(rc));
 636         }
 637
 638         /* pack the intent */
 639         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
 640         lit->opc = (__u64)it->it_op;
 641
 642         /* pack the layout intent request */
 643         layout = req_capsule_client_get(&req->rq_pill, &RMF_LAYOUT_INTENT);
 644         LASSERT(op_data->op_data != NULL);
 645         LASSERT(op_data->op_data_size == sizeof(*layout));
 646         memcpy(layout, op_data->op_data, sizeof(*layout));
 647
 648         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
 649                              obd->u.cli.cl_default_mds_easize);
 650         ptlrpc_request_set_replen(req);
 651         RETURN(req);
 652 }
 653
 654 static struct ptlrpc_request *mdc_enqueue_pack(struct obd_export *exp,
 655                                                int lvb_len)
 656 {
 657         struct ptlrpc_request *req;
 658         int rc;
 659
 660         ENTRY;
 661         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LDLM_ENQUEUE);
 662         if (req == NULL)
 663                 RETURN(ERR_PTR(-ENOMEM));
 664
 665         rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
 666         if (rc) {
 667                 ptlrpc_request_free(req);
 668                 RETURN(ERR_PTR(rc));
 669         }
 670
 671         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER, lvb_len);
 672         ptlrpc_request_set_replen(req);
 673         RETURN(req);
 674 }
 675
 676 static int mdc_finish_enqueue(struct obd_export *exp,
 677                               struct ptlrpc_request *req,
 678                               struct ldlm_enqueue_info *einfo,
 679                               struct lookup_intent *it,
 680                               struct lustre_handle *lockh, int rc)
 681 {
 682         struct req_capsule *pill = &req->rq_pill;
 683         struct ldlm_request *lockreq;
 684         struct ldlm_reply *lockrep;
 685         struct ldlm_lock *lock;
 686         struct mdt_body *body = NULL;
 687         void *lvb_data = NULL;
 688         __u32 lvb_len = 0;
 689
 690         ENTRY;
 691         LASSERT(rc >= 0);
 692         /* Similarly, if we're going to replay this request, we don't want to
 693          * actually get a lock, just perform the intent.
 694          */
 695         if (req->rq_transno || req->rq_replay) {
 696                 lockreq = req_capsule_client_get(pill, &RMF_DLM_REQ);
 697                 lockreq->lock_flags |= ldlm_flags_to_wire(LDLM_FL_INTENT_ONLY);
 698         }
 699
 700         if (rc == ELDLM_LOCK_ABORTED) {
 701                 einfo->ei_mode = 0;
 702                 memset(lockh, 0, sizeof(*lockh));
 703                 rc = 0;
 704         } else { /* rc = 0 */
 705                 lock = ldlm_handle2lock(lockh);
 706                 LASSERT(lock != NULL);
 707
 708                 /* If server returned a different lock mode, fix up variables */
 709                 if (lock->l_req_mode != einfo->ei_mode) {
 710                         ldlm_lock_addref(lockh, lock->l_req_mode);
 711                         ldlm_lock_decref(lockh, einfo->ei_mode);
 712                         einfo->ei_mode = lock->l_req_mode;
 713                 }
 714                 LDLM_LOCK_PUT(lock);
 715         }
 716
 717         lockrep = req_capsule_server_get(pill, &RMF_DLM_REP);
 718         LASSERT(lockrep != NULL); /* checked by ldlm_cli_enqueue() */
 719
 720         it->it_disposition = (int)lockrep->lock_policy_res1;
 721         it->it_status = (int)lockrep->lock_policy_res2;
 722         it->it_lock_mode = einfo->ei_mode;
 723         it->it_lock_handle = lockh->cookie;
 724         it->it_request = req;
 725
 726         /* Technically speaking rq_transno must already be zero if
 727          * it_status is in error, so the check is a bit redundant.
 728          */
 729         if ((!req->rq_transno || it->it_status < 0) && req->rq_replay)
 730                 mdc_clear_replay_flag(req, it->it_status);
 731
 732         /* If we're doing an IT_OPEN which did not result in an actual
 733          * successful open, then we need to remove the bit which saves
 734          * this request for unconditional replay.
 735          *
 736          * It's important that we do this first!  Otherwise we might exit the
 737          * function without doing so, and try to replay a failed create.
 738          * (b=3440)
 739          */
 740         if (it->it_op & IT_OPEN && req->rq_replay &&
 741             (!it_disposition(it, DISP_OPEN_OPEN) || it->it_status != 0))
 742                 mdc_clear_replay_flag(req, it->it_status);
 743
 744         DEBUG_REQ(D_RPCTRACE, req, "op=%x disposition=%x, status=%d",
 745                   it->it_op, it->it_disposition, it->it_status);
 746
 747         /* We know what to expect, so we do any byte flipping required here */
 748         if (it_has_reply_body(it)) {
 749                 body = req_capsule_server_get(pill, &RMF_MDT_BODY);
 750                 if (body == NULL) {
 751                         rc = -EPROTO;
 752                         CERROR("%s: cannot swab mdt_body: rc = %d\n",
 753                                exp->exp_obd->obd_name, rc);
 754                         RETURN(rc);
 755                 }
 756
 757                 if (it_disposition(it, DISP_OPEN_OPEN) &&
 758                     !it_open_error(DISP_OPEN_OPEN, it)) {
 759                         /*
 760                          * If this is a successful OPEN request, we need to set
 761                          * replay handler and data early, so that if replay
 762                          * happens immediately after swabbing below, new reply
 763                          * is swabbed by that handler correctly.
 764                          */
 765                         mdc_set_open_replay_data(NULL, NULL, it);
 766                 }
 767
 768                 if (it_disposition(it, DISP_OPEN_CREATE) &&
 769                     !it_open_error(DISP_OPEN_CREATE, it)) {
 770                         lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
 771                                              LPROC_MD_CREATE);
 772                 }
 773
 774                 if (body->mbo_valid & (OBD_MD_FLDIREA | OBD_MD_FLEASIZE)) {
 775                         void *eadata;
 776
 777                         mdc_update_max_ea_from_body(exp, body);
 778
 779                         /*
 780                          * The eadata is opaque; just check that it is there.
 781                          * Eventually, obd_unpackmd() will check the contents.
 782                          */
 783                         eadata = req_capsule_server_sized_get(pill, &RMF_MDT_MD,
 784                                                         body->mbo_eadatasize);
 785                         if (eadata == NULL)
 786                                 RETURN(-EPROTO);
 787
 788                         /* save LVB data and length if for layout lock */
 789                         lvb_data = eadata;
 790                         lvb_len = body->mbo_eadatasize;
 791
 792                         /*
 793                          * We save the reply LOV EA in case we have to replay a
 794                          * create for recovery.  If we didn't allocate a large
 795                          * enough request buffer above we need to reallocate it
 796                          * here to hold the actual LOV EA.
 797                          *
 798                          * To not save LOV EA if request is not going to replay
 799                          * (for example error one).
 800                          */
 801                         if ((it->it_op & IT_OPEN) && req->rq_replay) {
 802                                 rc = mdc_save_lovea(req, eadata,
 803                                                     body->mbo_eadatasize);
 804                                 if (rc) {
 805                                         body->mbo_valid &= ~OBD_MD_FLEASIZE;
 806                                         body->mbo_eadatasize = 0;
 807                                         rc = 0;
 808                                 }
 809                         }
 810                 }
 811         } else if (it->it_op & IT_LAYOUT) {
 812                 /* maybe the lock was granted right away and layout
 813                  * is packed into RMF_DLM_LVB of req
 814                  */
 815                 lvb_len = req_capsule_get_size(pill, &RMF_DLM_LVB, RCL_SERVER);
 816                 CDEBUG(D_INFO, "%s: layout return lvb %d transno %lld\n",
 817                        class_exp2obd(exp)->obd_name, lvb_len, req->rq_transno);
 818                 if (lvb_len > 0) {
 819                         lvb_data = req_capsule_server_sized_get(pill,
 820                                                         &RMF_DLM_LVB, lvb_len);
 821                         if (lvb_data == NULL)
 822                                 RETURN(-EPROTO);
 823
 824                         /**
 825                          * save replied layout data to the request buffer for
 826                          * recovery consideration (lest MDS reinitialize
 827                          * another set of OST objects).
 828                          */
 829                         if (req->rq_transno)
 830                                 (void)mdc_save_lovea(req, lvb_data, lvb_len);
 831                 }
 832         }
 833
 834         /* fill in stripe data for layout lock.
 835          * LU-6581: trust layout data only if layout lock is granted. The MDT
 836          * has stopped sending layout unless the layout lock is granted. The
 837          * client still does this checking in case it's talking with an old
 838          * server. - Jinshan
 839          */
 840         lock = ldlm_handle2lock(lockh);
 841         if (lock == NULL)
 842                 RETURN(rc);
 843
 844         if (ldlm_has_layout(lock) && lvb_data != NULL &&
 845             !(lockrep->lock_flags & LDLM_FL_BLOCKED_MASK)) {
 846                 void *lmm;
 847
 848                 LDLM_DEBUG(lock, "layout lock returned by: %s, lvb_len: %d",
 849                         ldlm_it2str(it->it_op), lvb_len);
 850
 851                 OBD_ALLOC_LARGE(lmm, lvb_len);
 852                 if (lmm == NULL)
 853                         GOTO(out_lock, rc = -ENOMEM);
 854
 855                 memcpy(lmm, lvb_data, lvb_len);
 856
 857                 /* install lvb_data */
 858                 lock_res_and_lock(lock);
 859                 if (lock->l_lvb_data == NULL) {
 860                         lock->l_lvb_type = LVB_T_LAYOUT;
 861                         lock->l_lvb_data = lmm;
 862                         lock->l_lvb_len = lvb_len;
 863                         lmm = NULL;
 864                 }
 865                 unlock_res_and_lock(lock);
 866                 if (lmm != NULL)
 867                         OBD_FREE_LARGE(lmm, lvb_len);
 868         }
 869
 870         if (ldlm_has_dom(lock)) {
 871                 LASSERT(lock->l_glimpse_ast == mdc_ldlm_glimpse_ast);
 872
 873                 body = req_capsule_server_get(pill, &RMF_MDT_BODY);
 874                 if (!(body->mbo_valid & OBD_MD_DOM_SIZE)) {
 875                         LDLM_ERROR(lock, "%s: DoM lock without size.",
 876                                    exp->exp_obd->obd_name);
 877                         GOTO(out_lock, rc = -EPROTO);
 878                 }
 879
 880                 LDLM_DEBUG(lock, "DoM lock is returned by: %s, size: %llu",
 881                            ldlm_it2str(it->it_op), body->mbo_dom_size);
 882
 883                 lock_res_and_lock(lock);
 884                 mdc_body2lvb(body, &lock->l_ost_lvb);
 885                 ldlm_lock_allow_match_locked(lock);
 886                 unlock_res_and_lock(lock);
 887         }
 888 out_lock:
 889         LDLM_LOCK_PUT(lock);
 890
 891         RETURN(rc);
 892 }
 893
 894 static inline bool mdc_skip_mod_rpc_slot(const struct lookup_intent *it)
 895 {
 896         if (it != NULL &&
 897             (it->it_op == IT_GETATTR || it->it_op == IT_LOOKUP ||
 898              it->it_op == IT_READDIR || it->it_op == IT_GETXATTR ||
 899              (it->it_op == IT_LAYOUT && !(it->it_flags & MDS_FMODE_WRITE))))
 900                 return true;
 901         return false;
 902 }
 903
 904 /* We always reserve enough space in the reply packet for a stripe MD, because
 905  * we don't know in advance the file type.
 906  */
 907 static int mdc_enqueue_base(struct obd_export *exp,
 908                             struct ldlm_enqueue_info *einfo,
 909                             const union ldlm_policy_data *policy,
 910                             struct lookup_intent *it,
 911                             struct md_op_data *op_data,
 912                             struct lustre_handle *lockh,
 913                             __u64 extra_lock_flags)
 914 {
 915         struct obd_device *obd = class_exp2obd(exp);
 916         struct ptlrpc_request *req;
 917         __u64 flags, saved_flags = extra_lock_flags;
 918         struct ldlm_res_id res_id;
 919         static const union ldlm_policy_data lookup_policy = {
 920                                   .l_inodebits = { MDS_INODELOCK_LOOKUP } };
 921         static const union ldlm_policy_data update_policy = {
 922                                   .l_inodebits = { MDS_INODELOCK_UPDATE } };
 923         static const union ldlm_policy_data layout_policy = {
 924                                   .l_inodebits = { MDS_INODELOCK_LAYOUT } };
 925         static const union ldlm_policy_data getxattr_policy = {
 926                                   .l_inodebits = { MDS_INODELOCK_XATTR } };
 927         int generation, resends = 0;
 928         struct ldlm_reply *lockrep;
 929         struct obd_import *imp = class_exp2cliimp(exp);
 930         __u32 acl_bufsize;
 931         enum lvb_type lvb_type = 0;
 932         int rc;
 933
 934         ENTRY;
 935         LASSERTF(!it || einfo->ei_type == LDLM_IBITS, "lock type %d\n",
 936                  einfo->ei_type);
 937         fid_build_reg_res_name(&op_data->op_fid1, &res_id);
 938
 939         if (it != NULL) {
 940                 LASSERT(policy == NULL);
 941
 942                 saved_flags |= LDLM_FL_HAS_INTENT;
 943                 if (it->it_op & (IT_GETATTR | IT_READDIR))
 944                         policy = &update_policy;
 945                 else if (it->it_op & IT_LAYOUT)
 946                         policy = &layout_policy;
 947                 else if (it->it_op & IT_GETXATTR)
 948                         policy = &getxattr_policy;
 949                 else
 950                         policy = &lookup_policy;
 951         }
 952
 953         generation = obd->u.cli.cl_import->imp_generation;
 954         if (!it || (it->it_op & (IT_OPEN | IT_CREAT)))
 955                 acl_bufsize = min_t(__u32, imp->imp_connect_data.ocd_max_easize,
 956                                     XATTR_SIZE_MAX);
 957         else
 958                 acl_bufsize = LUSTRE_POSIX_ACL_MAX_SIZE_OLD;
 959
 960 resend:
 961         flags = saved_flags;
 962         if (it == NULL) {
 963                 /* The only way right now is FLOCK. */
 964                 LASSERTF(einfo->ei_type == LDLM_FLOCK, "lock type %d\n",
 965                          einfo->ei_type);
 966                 res_id.name[3] = LDLM_FLOCK;
 967                 req = ldlm_enqueue_pack(exp, 0);
 968         } else if (it->it_op & IT_OPEN) {
 969                 req = mdc_intent_open_pack(exp, it, op_data, acl_bufsize);
 970         } else if (it->it_op & (IT_GETATTR | IT_LOOKUP)) {
 971                 req = mdc_intent_getattr_pack(exp, it, op_data, acl_bufsize);
 972         } else if (it->it_op & IT_READDIR) {
 973                 req = mdc_enqueue_pack(exp, 0);
 974         } else if (it->it_op & IT_LAYOUT) {
 975                 if (!imp_connect_lvb_type(imp))
 976                         RETURN(-EOPNOTSUPP);
 977                 req = mdc_intent_layout_pack(exp, it, op_data);
 978                 lvb_type = LVB_T_LAYOUT;
 979         } else if (it->it_op & IT_GETXATTR) {
 980                 req = mdc_intent_getxattr_pack(exp, it, op_data);
 981         } else {
 982                 LBUG();
 983                 RETURN(-EINVAL);
 984         }
 985
 986         if (IS_ERR(req))
 987                 RETURN(PTR_ERR(req));
 988
 989         if (resends) {
 990                 req->rq_generation_set = 1;
 991                 req->rq_import_generation = generation;
 992                 req->rq_sent = ktime_get_real_seconds() + resends;
 993         }
 994
 995         einfo->ei_req_slot = !(op_data->op_cli_flags & CLI_NO_SLOT);
 996         einfo->ei_mod_slot = !mdc_skip_mod_rpc_slot(it);
 997
 998         /* With Data-on-MDT the glimpse callback is needed too.
 999          * It is set here in advance but not in mdc_finish_enqueue()
1000          * to avoid possible races. It is safe to have glimpse handler
1001          * for non-DOM locks and costs nothing.
1002          */
1003         if (einfo->ei_cb_gl == NULL)
1004                 einfo->ei_cb_gl = mdc_ldlm_glimpse_ast;
1005
1006         rc = ldlm_cli_enqueue(exp, &req, einfo, &res_id, policy, &flags, NULL,
1007                               0, lvb_type, lockh, 0);
1008
1009         if (!it) {
1010                 /* For flock requests we immediatelly return without further
1011                  * delay and let caller deal with the rest, since rest of
1012                  * this function metadata processing makes no sense for flock
1013                  * requests anyway. But in case of problem during comms with
1014                  * server (-ETIMEDOUT) or any signal/kill attempt (-EINTR),
1015                  * we cannot rely on caller and this mainly for F_UNLCKs
1016                  * (explicits or automatically generated by kernel to clean
1017                  * current flocks upon exit) that can't be trashed.
1018                  */
1019                 ptlrpc_req_finished(req);
1020                 if (((rc == -EINTR) || (rc == -ETIMEDOUT)) &&
1021                     (einfo->ei_type == LDLM_FLOCK) &&
1022                     (einfo->ei_mode == LCK_NL))
1023                         goto resend;
1024                 RETURN(rc);
1025         }
1026
1027         if (rc < 0) {
1028                 CDEBUG(D_INFO,
1029                       "%s: ldlm_cli_enqueue "DFID":"DFID"=%s failed: rc = %d\n",
1030                       obd->obd_name, PFID(&op_data->op_fid1),
1031                       PFID(&op_data->op_fid2), op_data->op_name ?: "", rc);
1032
1033                 mdc_clear_replay_flag(req, rc);
1034                 ptlrpc_req_finished(req);
1035                 RETURN(rc);
1036         }
1037
1038         lockrep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
1039         LASSERT(lockrep != NULL);
1040
1041         lockrep->lock_policy_res2 =
1042                 ptlrpc_status_ntoh(lockrep->lock_policy_res2);
1043
1044         /* Retry infinitely when the server returns -EINPROGRESS for the
1045          * intent operation, when server returns -EINPROGRESS for acquiring
1046          * intent lock, we'll retry in after_reply().
1047          */
1048         if (it && (int)lockrep->lock_policy_res2 == -EINPROGRESS) {
1049                 mdc_clear_replay_flag(req, rc);
1050                 ptlrpc_req_finished(req);
1051                 if (generation == obd->u.cli.cl_import->imp_generation) {
1052                         if (signal_pending(current))
1053                                 RETURN(-EINTR);
1054
1055                         resends++;
1056                         CDEBUG(D_HA, "%s: resend:%d op:%d "DFID"/"DFID"\n",
1057                                obd->obd_name, resends, it->it_op,
1058                                PFID(&op_data->op_fid1),
1059                                PFID(&op_data->op_fid2));
1060                         goto resend;
1061                 } else {
1062                         CDEBUG(D_HA, "resend cross eviction\n");
1063                         RETURN(-EIO);
1064                 }
1065         }
1066
1067         if ((int)lockrep->lock_policy_res2 == -ERANGE &&
1068             it->it_op & (IT_OPEN | IT_GETATTR | IT_LOOKUP) &&
1069             acl_bufsize == LUSTRE_POSIX_ACL_MAX_SIZE_OLD) {
1070                 mdc_clear_replay_flag(req, -ERANGE);
1071                 ptlrpc_req_finished(req);
1072                 acl_bufsize = min_t(__u32, imp->imp_connect_data.ocd_max_easize,
1073                                     XATTR_SIZE_MAX);
1074                 goto resend;
1075         }
1076
1077         rc = mdc_finish_enqueue(exp, req, einfo, it, lockh, rc);
1078         if (rc < 0) {
1079                 if (lustre_handle_is_used(lockh)) {
1080                         ldlm_lock_decref(lockh, einfo->ei_mode);
1081                         memset(lockh, 0, sizeof(*lockh));
1082                 }
1083                 ptlrpc_req_finished(req);
1084
1085                 it->it_lock_handle = 0;
1086                 it->it_lock_mode = 0;
1087                 it->it_request = NULL;
1088         }
1089
1090         RETURN(rc);
1091 }
1092
1093 int mdc_enqueue(struct obd_export *exp, struct ldlm_enqueue_info *einfo,
1094                 const union ldlm_policy_data *policy,
1095                 struct md_op_data *op_data,
1096                 struct lustre_handle *lockh, __u64 extra_lock_flags)
1097 {
1098         return mdc_enqueue_base(exp, einfo, policy, NULL,
1099                                 op_data, lockh, extra_lock_flags);
1100 }
1101
1102 static int mdc_finish_intent_lock(struct obd_export *exp,
1103                                   struct ptlrpc_request *request,
1104                                   struct md_op_data *op_data,
1105                                   struct lookup_intent *it,
1106                                   struct lustre_handle *lockh)
1107 {
1108         struct lustre_handle old_lock;
1109         struct ldlm_lock *lock;
1110         int rc = 0;
1111
1112         ENTRY;
1113         LASSERT(request != NULL);
1114         LASSERT(request != LP_POISON);
1115         LASSERT(request->rq_repmsg != LP_POISON);
1116
1117         if (it->it_op & IT_READDIR)
1118                 RETURN(0);
1119
1120         if (it->it_op & (IT_GETXATTR | IT_LAYOUT)) {
1121                 if (it->it_status != 0)
1122                         GOTO(out, rc = it->it_status);
1123         } else {
1124                 if (!it_disposition(it, DISP_IT_EXECD)) {
1125                         /* The server failed before it even started executing
1126                          * the intent, i.e. because it couldn't unpack the
1127                          * request.
1128                          */
1129                         LASSERT(it->it_status != 0);
1130                         GOTO(out, rc = it->it_status);
1131                 }
1132                 rc = it_open_error(DISP_IT_EXECD, it);
1133                 if (rc)
1134                         GOTO(out, rc);
1135
1136                 rc = it_open_error(DISP_LOOKUP_EXECD, it);
1137                 if (rc)
1138                         GOTO(out, rc);
1139
1140                 /* keep requests around for the multiple phases of the call
1141                  * this shows the DISP_XX must guarantee we make it into the
1142                  * call
1143                  */
1144                 if (!it_disposition(it, DISP_ENQ_CREATE_REF) &&
1145                     it_disposition(it, DISP_OPEN_CREATE) &&
1146                     !it_open_error(DISP_OPEN_CREATE, it)) {
1147                         it_set_disposition(it, DISP_ENQ_CREATE_REF);
1148                         /* balanced in ll_create_node */
1149                         ptlrpc_request_addref(request);
1150                 }
1151                 if (!it_disposition(it, DISP_ENQ_OPEN_REF) &&
1152                     it_disposition(it, DISP_OPEN_OPEN) &&
1153                     !it_open_error(DISP_OPEN_OPEN, it)) {
1154                         it_set_disposition(it, DISP_ENQ_OPEN_REF);
1155                         /* balanced in ll_file_open */
1156                         ptlrpc_request_addref(request);
1157                         /* eviction in middle of open RPC processing b=11546 */
1158                         OBD_FAIL_TIMEOUT(OBD_FAIL_MDC_ENQUEUE_PAUSE,
1159                                          obd_timeout);
1160                 }
1161
1162                 if (it->it_op & IT_CREAT) {
1163                         /* XXX this belongs in ll_create_it */
1164                 } else if (it->it_op == IT_OPEN) {
1165                         LASSERT(!it_disposition(it, DISP_OPEN_CREATE));
1166                 } else {
1167                         LASSERT(it->it_op & (IT_GETATTR | IT_LOOKUP));
1168                 }
1169         }
1170
1171         /* If we already have a matching lock, then cancel the new
1172          * one.  We have to set the data here instead of in
1173          * mdc_enqueue, because we need to use the child's inode as
1174          * the l_ast_data to match, and that's not available until
1175          * intent_finish has performed the iget().
1176          */
1177         lock = ldlm_handle2lock(lockh);
1178         if (lock) {
1179                 union ldlm_policy_data policy = lock->l_policy_data;
1180
1181                 LDLM_DEBUG(lock, "matching against this");
1182
1183                 if (it_has_reply_body(it)) {
1184                         struct mdt_body *body;
1185
1186                         body = req_capsule_server_get(&request->rq_pill,
1187                                                       &RMF_MDT_BODY);
1188                         /* mdc_enqueue checked */
1189                         LASSERT(body != NULL);
1190                         LASSERTF(fid_res_name_eq(&body->mbo_fid1,
1191                                                  &lock->l_resource->lr_name),
1192                                  "Lock res_id: "DLDLMRES", fid: "DFID"\n",
1193                                  PLDLMRES(lock->l_resource),
1194                                  PFID(&body->mbo_fid1));
1195                 }
1196                 LDLM_LOCK_PUT(lock);
1197
1198                 memcpy(&old_lock, lockh, sizeof(*lockh));
1199                 if (ldlm_lock_match(NULL, LDLM_FL_BLOCK_GRANTED, NULL,
1200                                    LDLM_IBITS, &policy, LCK_NL, &old_lock)) {
1201                         ldlm_lock_decref_and_cancel(lockh, it->it_lock_mode);
1202                         memcpy(lockh, &old_lock, sizeof(old_lock));
1203                         it->it_lock_handle = lockh->cookie;
1204                 }
1205         }
1206
1207         EXIT;
1208 out:
1209         CDEBUG(D_DENTRY,
1210                "D_IT dentry=%.*s intent=%s status=%d disp=%x: rc = %d\n",
1211                 (int)op_data->op_namelen, op_data->op_name,
1212                 ldlm_it2str(it->it_op), it->it_status, it->it_disposition, rc);
1213
1214         return rc;
1215 }
1216
1217 int mdc_revalidate_lock(struct obd_export *exp, struct lookup_intent *it,
1218                         struct lu_fid *fid, __u64 *bits)
1219 {
1220         /* We could just return 1 immediately, but as we should only be called
1221          * in revalidate_it if we already have a lock, let's verify that.
1222          */
1223         struct ldlm_res_id res_id;
1224         struct lustre_handle lockh;
1225         union ldlm_policy_data policy;
1226         enum ldlm_mode mode;
1227
1228         ENTRY;
1229         if (it->it_lock_handle) {
1230                 lockh.cookie = it->it_lock_handle;
1231                 mode = ldlm_revalidate_lock_handle(&lockh, bits);
1232         } else {
1233                 fid_build_reg_res_name(fid, &res_id);
1234                 switch (it->it_op) {
1235                 case IT_GETATTR:
1236                         /* File attributes are held under multiple bits:
1237                          * nlink is under lookup lock, size and times are
1238                          * under UPDATE lock and recently we've also got
1239                          * a separate permissions lock for owner/group/acl that
1240                          * were protected by lookup lock before.
1241                          * Getattr must provide all of that information,
1242                          * so we need to ensure we have all of those locks.
1243                          * Unfortunately, if the bits are split across multiple
1244                          * locks, there's no easy way to match all of them here,
1245                          * so an extra RPC would be performed to fetch all
1246                          * of those bits at once for now.
1247                          */
1248                         /* For new MDTs(> 2.4), UPDATE|PERM should be enough,
1249                          * but for old MDTs (< 2.4), permission is covered
1250                          * by LOOKUP lock, so it needs to match all bits here.
1251                          */
1252                         policy.l_inodebits.bits = MDS_INODELOCK_UPDATE |
1253                                                   MDS_INODELOCK_PERM;
1254                         break;
1255                 case IT_READDIR:
1256                         policy.l_inodebits.bits = MDS_INODELOCK_UPDATE;
1257                         break;
1258                 case IT_LAYOUT:
1259                         policy.l_inodebits.bits = MDS_INODELOCK_LAYOUT;
1260                         break;
1261                 default:
1262                         policy.l_inodebits.bits = MDS_INODELOCK_LOOKUP;
1263                         break;
1264                 }
1265
1266                 mode = mdc_lock_match(exp, LDLM_FL_BLOCK_GRANTED, fid,
1267                                       LDLM_IBITS, &policy,
1268                                       LCK_CR | LCK_CW | LCK_PR | LCK_PW,
1269                                       &lockh);
1270         }
1271
1272         if (mode) {
1273                 it->it_lock_handle = lockh.cookie;
1274                 it->it_lock_mode = mode;
1275         } else {
1276                 it->it_lock_handle = 0;
1277                 it->it_lock_mode = 0;
1278         }
1279
1280         RETURN(!!mode);
1281 }
1282
1283 /*
1284  * This long block is all about fixing up the lock and request state
1285  * so that it is correct as of the moment _before_ the operation was
1286  * applied; that way, the VFS will think that everything is normal and
1287  * call Lustre's regular VFS methods.
1288  *
1289  * If we're performing a creation, that means that unless the creation
1290  * failed with EEXIST, we should fake up a negative dentry.
1291  *
1292  * For everything else, we want to lookup to succeed.
1293  *
1294  * One additional note: if CREATE or OPEN succeeded, we add an extra
1295  * reference to the request because we need to keep it around until
1296  * ll_create/ll_open gets called.
1297  *
1298  * The server will return to us, in it_disposition, an indication of
1299  * exactly what it_status refers to.
1300  *
1301  * If DISP_OPEN_OPEN is set, then it_status refers to the open() call,
1302  * otherwise if DISP_OPEN_CREATE is set, then it status is the
1303  * creation failure mode.  In either case, one of DISP_LOOKUP_NEG or
1304  * DISP_LOOKUP_POS will be set, indicating whether the child lookup
1305  * was successful.
1306  *
1307  * Else, if DISP_LOOKUP_EXECD then it_status is the rc of the
1308  * child lookup.
1309  */
1310 int mdc_intent_lock(struct obd_export *exp, struct md_op_data *op_data,
1311                     struct lookup_intent *it, struct ptlrpc_request **reqp,
1312                     ldlm_blocking_callback cb_blocking, __u64 extra_lock_flags)
1313 {
1314         struct ldlm_enqueue_info einfo = {
1315                 .ei_type        = LDLM_IBITS,
1316                 .ei_mode        = it_to_lock_mode(it),
1317                 .ei_cb_bl       = cb_blocking,
1318                 .ei_cb_cp       = ldlm_completion_ast,
1319                 .ei_cb_gl       = mdc_ldlm_glimpse_ast,
1320         };
1321         struct lustre_handle lockh;
1322         int rc = 0;
1323
1324         ENTRY;
1325         LASSERT(it);
1326         CDEBUG(D_DLMTRACE, "(name: %.*s,"DFID") in obj "DFID
1327                 ", intent: %s flags %#llo\n", (int)op_data->op_namelen,
1328                 op_data->op_name, PFID(&op_data->op_fid2),
1329                 PFID(&op_data->op_fid1), ldlm_it2str(it->it_op),
1330                 it->it_flags);
1331
1332         lockh.cookie = 0;
1333         /* MDS_FID_OP is not a revalidate case */
1334         if (fid_is_sane(&op_data->op_fid2) &&
1335             (it->it_op & (IT_LOOKUP | IT_GETATTR | IT_READDIR)) &&
1336             !(op_data->op_bias & MDS_FID_OP)) {
1337                 /* We could just return 1 immediately, but since we should only
1338                  * be called in revalidate_it if we already have a lock, let's
1339                  * verify that.
1340                  */
1341                 it->it_lock_handle = 0;
1342                 rc = mdc_revalidate_lock(exp, it, &op_data->op_fid2, NULL);
1343                 /* Only return failure if it was not GETATTR by cfid
1344                  * (from inode_revalidate()).
1345                  */
1346                 if (rc || op_data->op_namelen != 0)
1347                         RETURN(rc);
1348         }
1349
1350         /* For case if upper layer did not alloc fid, do it now. */
1351         if (!fid_is_sane(&op_data->op_fid2) && it->it_op & IT_CREAT) {
1352                 rc = mdc_fid_alloc(NULL, exp, &op_data->op_fid2, op_data);
1353                 if (rc < 0) {
1354                         CERROR("%s: cannot allocate new FID: rc=%d\n",
1355                                exp->exp_obd->obd_name, rc);
1356                         RETURN(rc);
1357                 }
1358         }
1359
1360         rc = mdc_enqueue_base(exp, &einfo, NULL, it, op_data, &lockh,
1361                               extra_lock_flags);
1362         if (rc < 0)
1363                 RETURN(rc);
1364
1365         *reqp = it->it_request;
1366         rc = mdc_finish_intent_lock(exp, *reqp, op_data, it, &lockh);
1367         RETURN(rc);
1368 }
1369
1370 static int mdc_intent_getattr_async_interpret(const struct lu_env *env,
1371                                               struct ptlrpc_request *req,
1372                                               void *args, int rc)
1373 {
1374         struct mdc_getattr_args *ga = args;
1375         struct obd_export *exp = ga->ga_exp;
1376         struct md_op_item *item = ga->ga_item;
1377         struct ldlm_enqueue_info *einfo = &item->mop_einfo;
1378         struct lookup_intent *it = &item->mop_it;
1379         struct lustre_handle *lockh = &item->mop_lockh;
1380         struct ldlm_reply *lockrep;
1381         __u64 flags = LDLM_FL_HAS_INTENT;
1382
1383         ENTRY;
1384         if (OBD_FAIL_CHECK(OBD_FAIL_MDC_GETATTR_ENQUEUE))
1385                 rc = -ETIMEDOUT;
1386
1387         rc = ldlm_cli_enqueue_fini(exp, req, einfo, 1, &flags, NULL, 0,
1388                                    lockh, rc, true);
1389         if (rc < 0) {
1390                 CERROR("%s: ldlm_cli_enqueue_fini() failed: rc = %d\n",
1391                        exp->exp_obd->obd_name, rc);
1392                 mdc_clear_replay_flag(req, rc);
1393                 GOTO(out, rc);
1394         }
1395
1396         lockrep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
1397         LASSERT(lockrep != NULL);
1398
1399         lockrep->lock_policy_res2 =
1400                 ptlrpc_status_ntoh(lockrep->lock_policy_res2);
1401
1402         rc = mdc_finish_enqueue(exp, req, einfo, it, lockh, rc);
1403         if (rc)
1404                 GOTO(out, rc);
1405
1406         rc = mdc_finish_intent_lock(exp, req, &item->mop_data, it, lockh);
1407         EXIT;
1408
1409 out:
1410         item->mop_cb(&req->rq_pill, item, rc);
1411         return 0;
1412 }
1413
1414 int mdc_intent_getattr_async(struct obd_export *exp,
1415                              struct md_op_item *item)
1416 {
1417         struct md_op_data *op_data = &item->mop_data;
1418         struct lookup_intent *it = &item->mop_it;
1419         struct ptlrpc_request *req;
1420         struct mdc_getattr_args *ga;
1421         struct ldlm_res_id res_id;
1422         union ldlm_policy_data policy = {
1423                 .l_inodebits = { MDS_INODELOCK_LOOKUP | MDS_INODELOCK_UPDATE }
1424         };
1425         __u64 flags = LDLM_FL_HAS_INTENT;
1426         int rc = 0;
1427
1428         ENTRY;
1429         CDEBUG(D_DLMTRACE,
1430                "name: %.*s in inode "DFID", intent: %s flags %#llo\n",
1431                (int)op_data->op_namelen, op_data->op_name,
1432                PFID(&op_data->op_fid1), ldlm_it2str(it->it_op), it->it_flags);
1433
1434         fid_build_reg_res_name(&op_data->op_fid1, &res_id);
1435         /* If the MDT return -ERANGE because of large ACL, then the sponsor
1436          * of the async getattr RPC will handle that by itself.
1437          */
1438         req = mdc_intent_getattr_pack(exp, it, op_data,
1439                                       LUSTRE_POSIX_ACL_MAX_SIZE_OLD);
1440         if (IS_ERR(req))
1441                 RETURN(PTR_ERR(req));
1442
1443         /* With Data-on-MDT the glimpse callback is needed too.
1444          * It is set here in advance but not in mdc_finish_enqueue()
1445          * to avoid possible races. It is safe to have glimpse handler
1446          * for non-DOM locks and costs nothing.
1447          */
1448         if (item->mop_einfo.ei_cb_gl == NULL)
1449                 item->mop_einfo.ei_cb_gl = mdc_ldlm_glimpse_ast;
1450
1451         rc = ldlm_cli_enqueue(exp, &req, &item->mop_einfo, &res_id, &policy,
1452                               &flags, NULL, 0, LVB_T_NONE, &item->mop_lockh, 1);
1453         if (rc < 0) {
1454                 ptlrpc_req_finished(req);
1455                 RETURN(rc);
1456         }
1457
1458         ga = ptlrpc_req_async_args(ga, req);
1459         ga->ga_exp = exp;
1460         ga->ga_item = item;
1461
1462         req->rq_interpret_reply = mdc_intent_getattr_async_interpret;
1463         ptlrpcd_add_req(req);
1464
1465         RETURN(0);
1466 }