drivers/staging/lustre/lustre/mdc/mdc_reint.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) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2015, 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 # include <linux/kernel.h>
  37
  38 #include <obd_class.h>
  39 #include "mdc_internal.h"
  40 #include <lustre_fid.h>
  41
  42 /* mdc_setattr does its own semaphore handling */
  43 static int mdc_reint(struct ptlrpc_request *request, int level)
  44 {
  45         int rc;
  46
  47         request->rq_send_state = level;
  48
  49         mdc_get_mod_rpc_slot(request, NULL);
  50         rc = ptlrpc_queue_wait(request);
  51         mdc_put_mod_rpc_slot(request, NULL);
  52         if (rc)
  53                 CDEBUG(D_INFO, "error in handling %d\n", rc);
  54         else if (!req_capsule_server_get(&request->rq_pill, &RMF_MDT_BODY))
  55                 rc = -EPROTO;
  56
  57         return rc;
  58 }
  59
  60 /* Find and cancel locally locks matched by inode @bits & @mode in the resource
  61  * found by @fid. Found locks are added into @cancel list. Returns the amount of
  62  * locks added to @cancels list.
  63  */
  64 int mdc_resource_get_unused(struct obd_export *exp, const struct lu_fid *fid,
  65                             struct list_head *cancels, enum ldlm_mode mode,
  66                             __u64 bits)
  67 {
  68         struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
  69         union ldlm_policy_data policy = {};
  70         struct ldlm_res_id res_id;
  71         struct ldlm_resource *res;
  72         int count;
  73
  74         /* Return, i.e. cancel nothing, only if ELC is supported (flag in
  75          * export) but disabled through procfs (flag in NS).
  76          *
  77          * This distinguishes from a case when ELC is not supported originally,
  78          * when we still want to cancel locks in advance and just cancel them
  79          * locally, without sending any RPC.
  80          */
  81         if (exp_connect_cancelset(exp) && !ns_connect_cancelset(ns))
  82                 return 0;
  83
  84         fid_build_reg_res_name(fid, &res_id);
  85         res = ldlm_resource_get(exp->exp_obd->obd_namespace,
  86                                 NULL, &res_id, 0, 0);
  87         if (IS_ERR(res))
  88                 return 0;
  89         LDLM_RESOURCE_ADDREF(res);
  90         /* Initialize ibits lock policy. */
  91         policy.l_inodebits.bits = bits;
  92         count = ldlm_cancel_resource_local(res, cancels, &policy,
  93                                            mode, 0, 0, NULL);
  94         LDLM_RESOURCE_DELREF(res);
  95         ldlm_resource_putref(res);
  96         return count;
  97 }
  98
  99 int mdc_setattr(struct obd_export *exp, struct md_op_data *op_data,
 100                 void *ea, size_t ealen, struct ptlrpc_request **request)
 101 {
 102         LIST_HEAD(cancels);
 103         struct ptlrpc_request *req;
 104         int count = 0, rc;
 105         __u64 bits;
 106
 107         bits = MDS_INODELOCK_UPDATE;
 108         if (op_data->op_attr.ia_valid & (ATTR_MODE | ATTR_UID | ATTR_GID))
 109                 bits |= MDS_INODELOCK_LOOKUP;
 110         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
 111             (fid_is_sane(&op_data->op_fid1)))
 112                 count = mdc_resource_get_unused(exp, &op_data->op_fid1,
 113                                                 &cancels, LCK_EX, bits);
 114         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 115                                    &RQF_MDS_REINT_SETATTR);
 116         if (!req) {
 117                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
 118                 return -ENOMEM;
 119         }
 120         req_capsule_set_size(&req->rq_pill, &RMF_MDT_EPOCH, RCL_CLIENT, 0);
 121         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT, ealen);
 122         req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES, RCL_CLIENT, 0);
 123
 124         rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
 125         if (rc) {
 126                 ptlrpc_request_free(req);
 127                 return rc;
 128         }
 129
 130         if (op_data->op_attr.ia_valid & (ATTR_MTIME | ATTR_CTIME))
 131                 CDEBUG(D_INODE, "setting mtime %ld, ctime %ld\n",
 132                        LTIME_S(op_data->op_attr.ia_mtime),
 133                        LTIME_S(op_data->op_attr.ia_ctime));
 134         mdc_setattr_pack(req, op_data, ea, ealen);
 135
 136         ptlrpc_request_set_replen(req);
 137
 138         rc = mdc_reint(req, LUSTRE_IMP_FULL);
 139
 140         if (rc == -ERESTARTSYS)
 141                 rc = 0;
 142
 143         *request = req;
 144
 145         return rc;
 146 }
 147
 148 int mdc_create(struct obd_export *exp, struct md_op_data *op_data,
 149                const void *data, size_t datalen, umode_t mode,
 150                uid_t uid, gid_t gid, cfs_cap_t cap_effective,
 151                __u64 rdev, struct ptlrpc_request **request)
 152 {
 153         struct ptlrpc_request *req;
 154         int level, rc;
 155         int count, resends = 0;
 156         struct obd_import *import = exp->exp_obd->u.cli.cl_import;
 157         int generation = import->imp_generation;
 158         LIST_HEAD(cancels);
 159
 160         /* For case if upper layer did not alloc fid, do it now. */
 161         if (!fid_is_sane(&op_data->op_fid2)) {
 162                 /*
 163                  * mdc_fid_alloc() may return errno 1 in case of switch to new
 164                  * sequence, handle this.
 165                  */
 166                 rc = mdc_fid_alloc(NULL, exp, &op_data->op_fid2, op_data);
 167                 if (rc < 0)
 168                         return rc;
 169         }
 170
 171 rebuild:
 172         count = 0;
 173         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
 174             (fid_is_sane(&op_data->op_fid1)))
 175                 count = mdc_resource_get_unused(exp, &op_data->op_fid1,
 176                                                 &cancels, LCK_EX,
 177                                                 MDS_INODELOCK_UPDATE);
 178
 179         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 180                                    &RQF_MDS_REINT_CREATE_ACL);
 181         if (!req) {
 182                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
 183                 return -ENOMEM;
 184         }
 185         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
 186                              op_data->op_namelen + 1);
 187         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT,
 188                              data && datalen ? datalen : 0);
 189
 190         rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
 191         if (rc) {
 192                 ptlrpc_request_free(req);
 193                 return rc;
 194         }
 195
 196         /*
 197          * mdc_create_pack() fills msg->bufs[1] with name and msg->bufs[2] with
 198          * tgt, for symlinks or lov MD data.
 199          */
 200         mdc_create_pack(req, op_data, data, datalen, mode, uid,
 201                         gid, cap_effective, rdev);
 202
 203         ptlrpc_request_set_replen(req);
 204
 205         /* ask ptlrpc not to resend on EINPROGRESS since we have our own retry
 206          * logic here
 207          */
 208         req->rq_no_retry_einprogress = 1;
 209
 210         if (resends) {
 211                 req->rq_generation_set = 1;
 212                 req->rq_import_generation = generation;
 213                 req->rq_sent = ktime_get_real_seconds() + resends;
 214         }
 215         level = LUSTRE_IMP_FULL;
 216  resend:
 217         rc = mdc_reint(req, level);
 218
 219         /* Resend if we were told to. */
 220         if (rc == -ERESTARTSYS) {
 221                 level = LUSTRE_IMP_RECOVER;
 222                 goto resend;
 223         } else if (rc == -EINPROGRESS) {
 224                 /* Retry create infinitely until succeed or get other
 225                  * error code.
 226                  */
 227                 ptlrpc_req_finished(req);
 228                 resends++;
 229
 230                 CDEBUG(D_HA, "%s: resend:%d create on " DFID "/" DFID "\n",
 231                        exp->exp_obd->obd_name, resends,
 232                        PFID(&op_data->op_fid1), PFID(&op_data->op_fid2));
 233
 234                 if (generation == import->imp_generation) {
 235                         goto rebuild;
 236                 } else {
 237                         CDEBUG(D_HA, "resend cross eviction\n");
 238                         return -EIO;
 239                 }
 240         }
 241
 242         *request = req;
 243         return rc;
 244 }
 245
 246 int mdc_unlink(struct obd_export *exp, struct md_op_data *op_data,
 247                struct ptlrpc_request **request)
 248 {
 249         LIST_HEAD(cancels);
 250         struct obd_device *obd = class_exp2obd(exp);
 251         struct ptlrpc_request *req = *request;
 252         int count = 0, rc;
 253
 254         LASSERT(!req);
 255
 256         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
 257             (fid_is_sane(&op_data->op_fid1)))
 258                 count = mdc_resource_get_unused(exp, &op_data->op_fid1,
 259                                                 &cancels, LCK_EX,
 260                                                 MDS_INODELOCK_UPDATE);
 261         if ((op_data->op_flags & MF_MDC_CANCEL_FID3) &&
 262             (fid_is_sane(&op_data->op_fid3)))
 263                 count += mdc_resource_get_unused(exp, &op_data->op_fid3,
 264                                                  &cancels, LCK_EX,
 265                                                  MDS_INODELOCK_FULL);
 266         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 267                                    &RQF_MDS_REINT_UNLINK);
 268         if (!req) {
 269                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
 270                 return -ENOMEM;
 271         }
 272         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
 273                              op_data->op_namelen + 1);
 274
 275         rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
 276         if (rc) {
 277                 ptlrpc_request_free(req);
 278                 return rc;
 279         }
 280
 281         mdc_unlink_pack(req, op_data);
 282
 283         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
 284                              obd->u.cli.cl_default_mds_easize);
 285         ptlrpc_request_set_replen(req);
 286
 287         *request = req;
 288
 289         rc = mdc_reint(req, LUSTRE_IMP_FULL);
 290         if (rc == -ERESTARTSYS)
 291                 rc = 0;
 292         return rc;
 293 }
 294
 295 int mdc_link(struct obd_export *exp, struct md_op_data *op_data,
 296              struct ptlrpc_request **request)
 297 {
 298         LIST_HEAD(cancels);
 299         struct ptlrpc_request *req;
 300         int count = 0, rc;
 301
 302         if ((op_data->op_flags & MF_MDC_CANCEL_FID2) &&
 303             (fid_is_sane(&op_data->op_fid2)))
 304                 count = mdc_resource_get_unused(exp, &op_data->op_fid2,
 305                                                 &cancels, LCK_EX,
 306                                                 MDS_INODELOCK_UPDATE);
 307         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
 308             (fid_is_sane(&op_data->op_fid1)))
 309                 count += mdc_resource_get_unused(exp, &op_data->op_fid1,
 310                                                  &cancels, LCK_EX,
 311                                                  MDS_INODELOCK_UPDATE);
 312
 313         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_REINT_LINK);
 314         if (!req) {
 315                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
 316                 return -ENOMEM;
 317         }
 318         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
 319                              op_data->op_namelen + 1);
 320
 321         rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
 322         if (rc) {
 323                 ptlrpc_request_free(req);
 324                 return rc;
 325         }
 326
 327         mdc_link_pack(req, op_data);
 328         ptlrpc_request_set_replen(req);
 329
 330         rc = mdc_reint(req, LUSTRE_IMP_FULL);
 331         *request = req;
 332         if (rc == -ERESTARTSYS)
 333                 rc = 0;
 334
 335         return rc;
 336 }
 337
 338 int mdc_rename(struct obd_export *exp, struct md_op_data *op_data,
 339                const char *old, size_t oldlen, const char *new, size_t newlen,
 340                struct ptlrpc_request **request)
 341 {
 342         LIST_HEAD(cancels);
 343         struct obd_device *obd = exp->exp_obd;
 344         struct ptlrpc_request *req;
 345         int count = 0, rc;
 346
 347         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
 348             (fid_is_sane(&op_data->op_fid1)))
 349                 count = mdc_resource_get_unused(exp, &op_data->op_fid1,
 350                                                 &cancels, LCK_EX,
 351                                                 MDS_INODELOCK_UPDATE);
 352         if ((op_data->op_flags & MF_MDC_CANCEL_FID2) &&
 353             (fid_is_sane(&op_data->op_fid2)))
 354                 count += mdc_resource_get_unused(exp, &op_data->op_fid2,
 355                                                  &cancels, LCK_EX,
 356                                                  MDS_INODELOCK_UPDATE);
 357         if ((op_data->op_flags & MF_MDC_CANCEL_FID3) &&
 358             (fid_is_sane(&op_data->op_fid3)))
 359                 count += mdc_resource_get_unused(exp, &op_data->op_fid3,
 360                                                  &cancels, LCK_EX,
 361                                                  MDS_INODELOCK_LOOKUP);
 362         if ((op_data->op_flags & MF_MDC_CANCEL_FID4) &&
 363             (fid_is_sane(&op_data->op_fid4)))
 364                 count += mdc_resource_get_unused(exp, &op_data->op_fid4,
 365                                                  &cancels, LCK_EX,
 366                                                  MDS_INODELOCK_FULL);
 367
 368         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 369                                    op_data->op_cli_flags & CLI_MIGRATE ?
 370                                    &RQF_MDS_REINT_MIGRATE : &RQF_MDS_REINT_RENAME);
 371         if (!req) {
 372                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
 373                 return -ENOMEM;
 374         }
 375
 376         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT, oldlen + 1);
 377         req_capsule_set_size(&req->rq_pill, &RMF_SYMTGT, RCL_CLIENT,
 378                              newlen + 1);
 379
 380         rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
 381         if (rc) {
 382                 ptlrpc_request_free(req);
 383                 return rc;
 384         }
 385
 386         if (op_data->op_cli_flags & CLI_MIGRATE && op_data->op_data) {
 387                 struct md_open_data *mod = op_data->op_data;
 388
 389                 LASSERTF(mod->mod_open_req &&
 390                          mod->mod_open_req->rq_type != LI_POISON,
 391                          "POISONED open %p!\n", mod->mod_open_req);
 392
 393                 DEBUG_REQ(D_HA, mod->mod_open_req, "matched open");
 394                 /*
 395                  * We no longer want to preserve this open for replay even
 396                  * though the open was committed. b=3632, b=3633
 397                  */
 398                 spin_lock(&mod->mod_open_req->rq_lock);
 399                 mod->mod_open_req->rq_replay = 0;
 400                 spin_unlock(&mod->mod_open_req->rq_lock);
 401         }
 402
 403         if (exp_connect_cancelset(exp) && req)
 404                 ldlm_cli_cancel_list(&cancels, count, req, 0);
 405
 406         mdc_rename_pack(req, op_data, old, oldlen, new, newlen);
 407
 408         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
 409                              obd->u.cli.cl_default_mds_easize);
 410         ptlrpc_request_set_replen(req);
 411
 412         rc = mdc_reint(req, LUSTRE_IMP_FULL);
 413         *request = req;
 414         if (rc == -ERESTARTSYS)
 415                 rc = 0;
 416
 417         return rc;
 418 }