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