lustre/mdc/mdc_reint.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  31  *
  32  * Copyright (c) 2012, Whamcloud, Inc.
  33  */
  34 /*
  35  * This file is part of Lustre, http://www.lustre.org/
  36  * Lustre is a trademark of Sun Microsystems, Inc.
  37  */
  38
  39 #ifndef EXPORT_SYMTAB
  40 # define EXPORT_SYMTAB
  41 #endif
  42 #define DEBUG_SUBSYSTEM S_MDC
  43
  44 #ifdef __KERNEL__
  45 #ifndef AUTOCONF_INCLUDED
  46 # include <linux/config.h>
  47 #endif
  48 # include <linux/module.h>
  49 # include <linux/kernel.h>
  50 #else
  51 # include <liblustre.h>
  52 #endif
  53
  54 #include <obd_class.h>
  55 #include "mdc_internal.h"
  56 #include <lustre_fid.h>
  57
  58 /* mdc_setattr does its own semaphore handling */
  59 static int mdc_reint(struct ptlrpc_request *request,
  60                      struct mdc_rpc_lock *rpc_lock,
  61                      int level)
  62 {
  63         int rc;
  64
  65         request->rq_send_state = level;
  66
  67         mdc_get_rpc_lock(rpc_lock, NULL);
  68         rc = ptlrpc_queue_wait(request);
  69         mdc_put_rpc_lock(rpc_lock, NULL);
  70         if (rc)
  71                 CDEBUG(D_INFO, "error in handling %d\n", rc);
  72         else if (!req_capsule_server_get(&request->rq_pill, &RMF_MDT_BODY)) {
  73                 rc = -EPROTO;
  74         }
  75         return rc;
  76 }
  77
  78 /* Find and cancel locally locks matched by inode @bits & @mode in the resource
  79  * found by @fid. Found locks are added into @cancel list. Returns the amount of
  80  * locks added to @cancels list. */
  81 int mdc_resource_get_unused(struct obd_export *exp, struct lu_fid *fid,
  82                             cfs_list_t *cancels, ldlm_mode_t mode,
  83                             __u64 bits)
  84 {
  85         ldlm_policy_data_t policy = {{0}};
  86         struct ldlm_res_id res_id;
  87         struct ldlm_resource *res;
  88         int count;
  89         ENTRY;
  90
  91         fid_build_reg_res_name(fid, &res_id);
  92         res = ldlm_resource_get(exp->exp_obd->obd_namespace,
  93                                 NULL, &res_id, 0, 0);
  94         if (res == NULL)
  95                 RETURN(0);
  96         LDLM_RESOURCE_ADDREF(res);
  97         /* Initialize ibits lock policy. */
  98         policy.l_inodebits.bits = bits;
  99         count = ldlm_cancel_resource_local(res, cancels, &policy,
 100                                            mode, 0, 0, NULL);
 101         LDLM_RESOURCE_DELREF(res);
 102         ldlm_resource_putref(res);
 103         RETURN(count);
 104 }
 105
 106 static int mdc_prep_elc_req(struct obd_export *exp, struct ptlrpc_request *req,
 107                             cfs_list_t *cancels, int count)
 108 {
 109         return ldlm_prep_elc_req(exp, req, LUSTRE_MDS_VERSION, MDS_REINT,
 110                                  0, cancels, count);
 111 }
 112
 113 int mdc_setattr(struct obd_export *exp, struct md_op_data *op_data,
 114                 void *ea, int ealen, void *ea2, int ea2len,
 115                 struct ptlrpc_request **request, struct md_open_data **mod)
 116 {
 117         CFS_LIST_HEAD(cancels);
 118         struct ptlrpc_request *req;
 119         struct mdc_rpc_lock *rpc_lock;
 120         struct obd_device *obd = exp->exp_obd;
 121         int count = 0, rc;
 122         __u64 bits;
 123         ENTRY;
 124
 125         LASSERT(op_data != NULL);
 126
 127         bits = MDS_INODELOCK_UPDATE;
 128         if (op_data->op_attr.ia_valid & (ATTR_MODE|ATTR_UID|ATTR_GID))
 129                 bits |= MDS_INODELOCK_LOOKUP;
 130         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
 131             (fid_is_sane(&op_data->op_fid1)) &&
 132             !OBD_FAIL_CHECK(OBD_FAIL_LDLM_BL_CALLBACK))
 133                 count = mdc_resource_get_unused(exp, &op_data->op_fid1,
 134                                                 &cancels, LCK_EX, bits);
 135         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 136                                    &RQF_MDS_REINT_SETATTR);
 137         if (req == NULL) {
 138                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
 139                 RETURN(-ENOMEM);
 140         }
 141         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
 142         if ((op_data->op_flags & (MF_SOM_CHANGE | MF_EPOCH_OPEN)) == 0)
 143                 req_capsule_set_size(&req->rq_pill, &RMF_MDT_EPOCH, RCL_CLIENT,
 144                                      0);
 145         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT, ealen);
 146         req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES, RCL_CLIENT,
 147                              ea2len);
 148
 149         rc = mdc_prep_elc_req(exp, req, &cancels, count);
 150         if (rc) {
 151                 ptlrpc_request_free(req);
 152                 RETURN(rc);
 153         }
 154
 155         rpc_lock = obd->u.cli.cl_rpc_lock;
 156
 157         if (op_data->op_attr.ia_valid & (ATTR_MTIME | ATTR_CTIME))
 158                 CDEBUG(D_INODE, "setting mtime "CFS_TIME_T
 159                        ", ctime "CFS_TIME_T"\n",
 160                        LTIME_S(op_data->op_attr.ia_mtime),
 161                        LTIME_S(op_data->op_attr.ia_ctime));
 162         mdc_setattr_pack(req, op_data, ea, ealen, ea2, ea2len);
 163
 164         ptlrpc_request_set_replen(req);
 165         if (mod && (op_data->op_flags & MF_EPOCH_OPEN) &&
 166             req->rq_import->imp_replayable)
 167         {
 168                 LASSERT(*mod == NULL);
 169
 170                 *mod = obd_mod_alloc();
 171                 if (*mod == NULL) {
 172                         DEBUG_REQ(D_ERROR, req, "Can't allocate "
 173                                   "md_open_data");
 174                 } else {
 175                         req->rq_replay = 1;
 176                         req->rq_cb_data = *mod;
 177                         (*mod)->mod_open_req = req;
 178                         req->rq_commit_cb = mdc_commit_open;
 179                         /**
 180                          * Take an extra reference on \var mod, it protects \var
 181                          * mod from being freed on eviction (commit callback is
 182                          * called despite rq_replay flag).
 183                          * Will be put on mdc_done_writing().
 184                          */
 185                         obd_mod_get(*mod);
 186                 }
 187         }
 188
 189         rc = mdc_reint(req, rpc_lock, LUSTRE_IMP_FULL);
 190
 191         /* Save the obtained info in the original RPC for the replay case. */
 192         if (rc == 0 && (op_data->op_flags & MF_EPOCH_OPEN)) {
 193                 struct mdt_ioepoch *epoch;
 194                 struct mdt_body  *body;
 195
 196                 epoch = req_capsule_client_get(&req->rq_pill, &RMF_MDT_EPOCH);
 197                 body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 198                 LASSERT(epoch != NULL);
 199                 LASSERT(body != NULL);
 200                 epoch->handle = body->handle;
 201                 epoch->ioepoch = body->ioepoch;
 202                 req->rq_replay_cb = mdc_replay_open;
 203         /** bug 3633, open may be committed and estale answer is not error */
 204         } else if (rc == -ESTALE && (op_data->op_flags & MF_SOM_CHANGE)) {
 205                 rc = 0;
 206         } else if (rc == -ERESTARTSYS) {
 207                 rc = 0;
 208         }
 209         *request = req;
 210         if (rc && req->rq_commit_cb) {
 211                 /* Put an extra reference on \var mod on error case. */
 212                 obd_mod_put(*mod);
 213                 req->rq_commit_cb(req);
 214         }
 215         RETURN(rc);
 216 }
 217
 218 int mdc_create(struct obd_export *exp, struct md_op_data *op_data,
 219                const void *data, int datalen, int mode, __u32 uid, __u32 gid,
 220                cfs_cap_t cap_effective, __u64 rdev,
 221                struct ptlrpc_request **request)
 222 {
 223         struct ptlrpc_request *req;
 224         int level, rc;
 225         int count = 0;
 226         CFS_LIST_HEAD(cancels);
 227         ENTRY;
 228
 229         /* For case if upper layer did not alloc fid, do it now. */
 230         if (!fid_is_sane(&op_data->op_fid2)) {
 231                 /*
 232                  * mdc_fid_alloc() may return errno 1 in case of switch to new
 233                  * sequence, handle this.
 234                  */
 235                 rc = mdc_fid_alloc(exp, &op_data->op_fid2, op_data);
 236                 if (rc < 0) {
 237                         CERROR("Can't alloc new fid, rc %d\n", rc);
 238                         RETURN(rc);
 239                 }
 240         }
 241
 242         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
 243             (fid_is_sane(&op_data->op_fid1)))
 244                 count = mdc_resource_get_unused(exp, &op_data->op_fid1,
 245                                                 &cancels, LCK_EX,
 246                                                 MDS_INODELOCK_UPDATE);
 247
 248         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 249                                    &RQF_MDS_REINT_CREATE_RMT_ACL);
 250         if (req == NULL) {
 251                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
 252                 RETURN(-ENOMEM);
 253         }
 254         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
 255         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
 256                              op_data->op_namelen + 1);
 257         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT,
 258                              data && datalen ? datalen : 0);
 259
 260         rc = mdc_prep_elc_req(exp, req, &cancels, count);
 261         if (rc) {
 262                 ptlrpc_request_free(req);
 263                 RETURN(rc);
 264         }
 265
 266         /*
 267          * mdc_create_pack() fills msg->bufs[1] with name and msg->bufs[2] with
 268          * tgt, for symlinks or lov MD data.
 269          */
 270         mdc_create_pack(req, op_data, data, datalen, mode, uid,
 271                         gid, cap_effective, rdev);
 272
 273         ptlrpc_request_set_replen(req);
 274
 275         level = LUSTRE_IMP_FULL;
 276  resend:
 277         rc = mdc_reint(req, exp->exp_obd->u.cli.cl_rpc_lock, level);
 278
 279         /* Resend if we were told to. */
 280         if (rc == -ERESTARTSYS) {
 281                 level = LUSTRE_IMP_RECOVER;
 282                 goto resend;
 283         } else if (rc == 0) {
 284                 struct mdt_body *body;
 285                 struct lustre_capa *capa;
 286
 287                 body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 288                 LASSERT(body);
 289                 if (body->valid & OBD_MD_FLMDSCAPA) {
 290                         capa = req_capsule_server_get(&req->rq_pill,
 291                                                       &RMF_CAPA1);
 292                         if (capa == NULL)
 293                                 rc = -EPROTO;
 294                 }
 295         }
 296
 297         *request = req;
 298         RETURN(rc);
 299 }
 300
 301 int mdc_unlink(struct obd_export *exp, struct md_op_data *op_data,
 302                struct ptlrpc_request **request)
 303 {
 304         CFS_LIST_HEAD(cancels);
 305         struct obd_device *obd = class_exp2obd(exp);
 306         struct ptlrpc_request *req = *request;
 307         int count = 0, rc;
 308         ENTRY;
 309
 310         LASSERT(req == NULL);
 311
 312         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
 313             (fid_is_sane(&op_data->op_fid1)) &&
 314             !OBD_FAIL_CHECK(OBD_FAIL_LDLM_BL_CALLBACK))
 315                 count = mdc_resource_get_unused(exp, &op_data->op_fid1,
 316                                                 &cancels, LCK_EX,
 317                                                 MDS_INODELOCK_UPDATE);
 318         if ((op_data->op_flags & MF_MDC_CANCEL_FID3) &&
 319             (fid_is_sane(&op_data->op_fid3)) &&
 320             !OBD_FAIL_CHECK(OBD_FAIL_LDLM_BL_CALLBACK))
 321                 count += mdc_resource_get_unused(exp, &op_data->op_fid3,
 322                                                  &cancels, LCK_EX,
 323                                                  MDS_INODELOCK_FULL);
 324         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 325                                    &RQF_MDS_REINT_UNLINK);
 326         if (req == NULL) {
 327                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
 328                 RETURN(-ENOMEM);
 329         }
 330         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
 331         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
 332                              op_data->op_namelen + 1);
 333
 334         rc = mdc_prep_elc_req(exp, req, &cancels, count);
 335         if (rc) {
 336                 ptlrpc_request_free(req);
 337                 RETURN(rc);
 338         }
 339
 340         mdc_unlink_pack(req, op_data);
 341
 342         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
 343                              obd->u.cli.cl_max_mds_easize);
 344         req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES, RCL_SERVER,
 345                              obd->u.cli.cl_max_mds_cookiesize);
 346         ptlrpc_request_set_replen(req);
 347
 348         *request = req;
 349
 350         rc = mdc_reint(req, obd->u.cli.cl_rpc_lock, LUSTRE_IMP_FULL);
 351         if (rc == -ERESTARTSYS)
 352                 rc = 0;
 353         RETURN(rc);
 354 }
 355
 356 int mdc_link(struct obd_export *exp, struct md_op_data *op_data,
 357              struct ptlrpc_request **request)
 358 {
 359         CFS_LIST_HEAD(cancels);
 360         struct obd_device *obd = exp->exp_obd;
 361         struct ptlrpc_request *req;
 362         int count = 0, rc;
 363         ENTRY;
 364
 365         if ((op_data->op_flags & MF_MDC_CANCEL_FID2) &&
 366             (fid_is_sane(&op_data->op_fid2)))
 367                 count = mdc_resource_get_unused(exp, &op_data->op_fid2,
 368                                                 &cancels, LCK_EX,
 369                                                 MDS_INODELOCK_UPDATE);
 370         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
 371             (fid_is_sane(&op_data->op_fid1)))
 372                 count += mdc_resource_get_unused(exp, &op_data->op_fid1,
 373                                                  &cancels, LCK_EX,
 374                                                  MDS_INODELOCK_UPDATE);
 375
 376         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_REINT_LINK);
 377         if (req == NULL) {
 378                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
 379                 RETURN(-ENOMEM);
 380         }
 381         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
 382         mdc_set_capa_size(req, &RMF_CAPA2, op_data->op_capa2);
 383         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
 384                              op_data->op_namelen + 1);
 385
 386         rc = mdc_prep_elc_req(exp, req, &cancels, count);
 387         if (rc) {
 388                 ptlrpc_request_free(req);
 389                 RETURN(rc);
 390         }
 391
 392         mdc_link_pack(req, op_data);
 393         ptlrpc_request_set_replen(req);
 394
 395         rc = mdc_reint(req, obd->u.cli.cl_rpc_lock, LUSTRE_IMP_FULL);
 396         *request = req;
 397         if (rc == -ERESTARTSYS)
 398                 rc = 0;
 399
 400         RETURN(rc);
 401 }
 402
 403 int mdc_rename(struct obd_export *exp, struct md_op_data *op_data,
 404                const char *old, int oldlen, const char *new, int newlen,
 405                struct ptlrpc_request **request)
 406 {
 407         CFS_LIST_HEAD(cancels);
 408         struct obd_device *obd = exp->exp_obd;
 409         struct ptlrpc_request *req;
 410         int count = 0, rc;
 411         ENTRY;
 412
 413         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
 414             (fid_is_sane(&op_data->op_fid1)))
 415                 count = mdc_resource_get_unused(exp, &op_data->op_fid1,
 416                                                 &cancels, LCK_EX,
 417                                                 MDS_INODELOCK_UPDATE);
 418         if ((op_data->op_flags & MF_MDC_CANCEL_FID2) &&
 419             (fid_is_sane(&op_data->op_fid2)))
 420                 count += mdc_resource_get_unused(exp, &op_data->op_fid2,
 421                                                  &cancels, LCK_EX,
 422                                                  MDS_INODELOCK_UPDATE);
 423         if ((op_data->op_flags & MF_MDC_CANCEL_FID3) &&
 424             (fid_is_sane(&op_data->op_fid3)))
 425                 count += mdc_resource_get_unused(exp, &op_data->op_fid3,
 426                                                  &cancels, LCK_EX,
 427                                                  MDS_INODELOCK_LOOKUP);
 428         if ((op_data->op_flags & MF_MDC_CANCEL_FID4) &&
 429              (fid_is_sane(&op_data->op_fid4)))
 430                 count += mdc_resource_get_unused(exp, &op_data->op_fid4,
 431                                                  &cancels, LCK_EX,
 432                                                  MDS_INODELOCK_FULL);
 433
 434         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 435                                    &RQF_MDS_REINT_RENAME);
 436         if (req == NULL) {
 437                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
 438                 RETURN(-ENOMEM);
 439         }
 440
 441         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
 442         mdc_set_capa_size(req, &RMF_CAPA2, op_data->op_capa2);
 443         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT, oldlen + 1);
 444         req_capsule_set_size(&req->rq_pill, &RMF_SYMTGT, RCL_CLIENT, newlen+1);
 445
 446         rc = mdc_prep_elc_req(exp, req, &cancels, count);
 447         if (rc) {
 448                 ptlrpc_request_free(req);
 449                 RETURN(rc);
 450         }
 451
 452         if (exp_connect_cancelset(exp) && req)
 453                 ldlm_cli_cancel_list(&cancels, count, req, 0);
 454
 455         mdc_rename_pack(req, op_data, old, oldlen, new, newlen);
 456
 457         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
 458                              obd->u.cli.cl_max_mds_easize);
 459         req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES, RCL_SERVER,
 460                              obd->u.cli.cl_max_mds_cookiesize);
 461         ptlrpc_request_set_replen(req);
 462
 463         rc = mdc_reint(req, obd->u.cli.cl_rpc_lock, LUSTRE_IMP_FULL);
 464         *request = req;
 465         if (rc == -ERESTARTSYS)
 466                 rc = 0;
 467
 468         RETURN(rc);
 469 }