lustre/mdt/mdt_reint.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  linux/mdt/mdt_reint.c
   5  *  Lustre Metadata Target (mdt) reintegration routines
   6  *
   7  *  Copyright (C) 2002-2006 Cluster File Systems, Inc.
   8  *   Author: Peter Braam <braam@clusterfs.com>
   9  *   Author: Andreas Dilger <adilger@clusterfs.com>
  10  *   Author: Phil Schwan <phil@clusterfs.com>
  11  *   Author: Huang Hua <huanghua@clusterfs.com>
  12  *   Author: Yury Umanets <umka@clusterfs.com>
  13  *
  14  *   This file is part of the Lustre file system, http://www.lustre.org
  15  *   Lustre is a trademark of Cluster File Systems, Inc.
  16  *
  17  *   You may have signed or agreed to another license before downloading
  18  *   this software.  If so, you are bound by the terms and conditions
  19  *   of that agreement, and the following does not apply to you.  See the
  20  *   LICENSE file included with this distribution for more information.
  21  *
  22  *   If you did not agree to a different license, then this copy of Lustre
  23  *   is open source software; you can redistribute it and/or modify it
  24  *   under the terms of version 2 of the GNU General Public License as
  25  *   published by the Free Software Foundation.
  26  *
  27  *   In either case, Lustre is distributed in the hope that it will be
  28  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  29  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  30  *   license text for more details.
  31  */
  32
  33 #ifndef EXPORT_SYMTAB
  34 # define EXPORT_SYMTAB
  35 #endif
  36 #define DEBUG_SUBSYSTEM S_MDS
  37
  38 #include "mdt_internal.h"
  39 #include "lu_time.h"
  40
  41 static inline void mdt_reint_init_ma(struct mdt_thread_info *info,
  42                                      struct md_attr *ma)
  43 {
  44         ma->ma_lmm = req_capsule_server_get(info->mti_pill, &RMF_MDT_MD);
  45         ma->ma_lmm_size = req_capsule_get_size(info->mti_pill,
  46                                                &RMF_MDT_MD, RCL_SERVER);
  47
  48         ma->ma_cookie = req_capsule_server_get(info->mti_pill,
  49                                                &RMF_LOGCOOKIES);
  50         ma->ma_cookie_size = req_capsule_get_size(info->mti_pill,
  51                                                   &RMF_LOGCOOKIES,
  52                                                   RCL_SERVER);
  53
  54         ma->ma_need = MA_INODE | MA_LOV | MA_COOKIE;
  55         ma->ma_valid = 0;
  56 }
  57
  58 static int mdt_create_pack_capa(struct mdt_thread_info *info, int rc,
  59                                 struct mdt_object *object,
  60                                 struct mdt_body *repbody)
  61 {
  62         ENTRY;
  63
  64         /* for cross-ref mkdir, mds capa has been fetched from remote obj, then
  65          * we won't go to below*/
  66         if (repbody->valid & OBD_MD_FLMDSCAPA)
  67                 RETURN(rc);
  68
  69         if (rc == 0 && info->mti_mdt->mdt_opts.mo_mds_capa) {
  70                 struct lustre_capa *capa;
  71
  72                 capa = req_capsule_server_get(info->mti_pill, &RMF_CAPA1);
  73                 LASSERT(capa);
  74                 capa->lc_opc = CAPA_OPC_MDS_DEFAULT;
  75                 rc = mo_capa_get(info->mti_env, mdt_object_child(object), capa,
  76                                  0);
  77                 if (rc == 0)
  78                         repbody->valid |= OBD_MD_FLMDSCAPA;
  79         }
  80
  81         RETURN(rc);
  82 }
  83
  84 static int mdt_md_create(struct mdt_thread_info *info)
  85 {
  86         struct mdt_device       *mdt = info->mti_mdt;
  87         struct mdt_object       *parent;
  88         struct mdt_object       *child;
  89         struct mdt_lock_handle  *lh;
  90         struct mdt_body         *repbody;
  91         struct md_attr          *ma = &info->mti_attr;
  92         struct mdt_reint_record *rr = &info->mti_rr;
  93         struct lu_name          *lname;
  94         int rc;
  95         ENTRY;
  96
  97         DEBUG_REQ(D_INODE, mdt_info_req(info), "Create  (%s->"DFID") in "DFID,
  98                   rr->rr_name, PFID(rr->rr_fid2), PFID(rr->rr_fid1));
  99
 100         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
 101
 102         lh = &info->mti_lh[MDT_LH_PARENT];
 103         mdt_lock_pdo_init(lh, LCK_PW, rr->rr_name, rr->rr_namelen);
 104
 105         parent = mdt_object_find_lock(info, rr->rr_fid1, lh,
 106                                       MDS_INODELOCK_UPDATE);
 107         if (IS_ERR(parent))
 108                 RETURN(PTR_ERR(parent));
 109
 110         child = mdt_object_find(info->mti_env, mdt, rr->rr_fid2);
 111         if (likely(!IS_ERR(child))) {
 112                 struct md_object *next = mdt_object_child(parent);
 113
 114                 ma->ma_need = MA_INODE;
 115                 ma->ma_valid = 0;
 116                 /* capa for cross-ref will be stored here */
 117                 ma->ma_capa = req_capsule_server_get(info->mti_pill,
 118                                                      &RMF_CAPA1);
 119                 LASSERT(ma->ma_capa);
 120
 121                 mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
 122                                OBD_FAIL_MDS_REINT_CREATE_WRITE);
 123
 124                 /* Let lower layer know current lock mode. */
 125                 info->mti_spec.sp_cr_mode =
 126                         mdt_dlm_mode2mdl_mode(lh->mlh_pdo_mode);
 127
 128                 /*
 129                  * Do perform lookup sanity check. We do not know if name exists
 130                  * or not.
 131                  */
 132                 info->mti_spec.sp_cr_lookup = 1;
 133
 134                 lname = mdt_name(info->mti_env, (char *)rr->rr_name,
 135                                  rr->rr_namelen);
 136                 rc = mdo_create(info->mti_env, next, lname,
 137                                 mdt_object_child(child),
 138                                 &info->mti_spec, ma);
 139                 if (rc == 0) {
 140                         /* Return fid & attr to client. */
 141                         if (ma->ma_valid & MA_INODE)
 142                                 mdt_pack_attr2body(info, repbody, &ma->ma_attr,
 143                                                    mdt_object_fid(child));
 144                 }
 145                 mdt_object_put(info->mti_env, child);
 146         } else
 147                 rc = PTR_ERR(child);
 148
 149         mdt_create_pack_capa(info, rc, child, repbody);
 150         mdt_object_unlock_put(info, parent, lh, rc);
 151         RETURN(rc);
 152 }
 153
 154 /* Partial request to create object only */
 155 static int mdt_md_mkobj(struct mdt_thread_info *info)
 156 {
 157         struct mdt_device      *mdt = info->mti_mdt;
 158         struct mdt_object      *o;
 159         struct mdt_body        *repbody;
 160         struct md_attr         *ma = &info->mti_attr;
 161         int rc;
 162         ENTRY;
 163
 164         DEBUG_REQ(D_INODE, mdt_info_req(info), "Partial create "DFID"",
 165                   PFID(info->mti_rr.rr_fid2));
 166
 167         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
 168
 169         o = mdt_object_find(info->mti_env, mdt, info->mti_rr.rr_fid2);
 170         if (!IS_ERR(o)) {
 171                 struct md_object *next = mdt_object_child(o);
 172
 173                 ma->ma_need = MA_INODE;
 174                 ma->ma_valid = 0;
 175
 176                 /*
 177                  * Cross-ref create can encounter already created obj in case of
 178                  * recovery, just get attr in that case.
 179                  */
 180                 if (mdt_object_exists(o) == 1) {
 181                         rc = mo_attr_get(info->mti_env, next, ma);
 182                 } else {
 183                         /*
 184                          * Here, NO permission check for object_create,
 185                          * such check has been done on the original MDS.
 186                          */
 187                         rc = mo_object_create(info->mti_env, next,
 188                                               &info->mti_spec, ma);
 189                 }
 190                 if (rc == 0) {
 191                         /* Return fid & attr to client. */
 192                         if (ma->ma_valid & MA_INODE)
 193                                 mdt_pack_attr2body(info, repbody, &ma->ma_attr,
 194                                                    mdt_object_fid(o));
 195                 }
 196                 mdt_object_put(info->mti_env, o);
 197         } else
 198                 rc = PTR_ERR(o);
 199
 200         mdt_create_pack_capa(info, rc, o, repbody);
 201         RETURN(rc);
 202 }
 203
 204 /* In the raw-setattr case, we lock the child inode.
 205  * In the write-back case or if being called from open,
 206  *               the client holds a lock already.
 207  * We use the ATTR_FROM_OPEN (translated into MRF_SETATTR_LOCKED by
 208  * mdt_setattr_unpack()) flag to tell these cases apart. */
 209 int mdt_attr_set(struct mdt_thread_info *info, struct mdt_object *mo, int flags)
 210 {
 211         struct md_attr          *ma = &info->mti_attr;
 212         struct mdt_lock_handle  *lh;
 213         int som_update = 0;
 214         int rc;
 215         ENTRY;
 216
 217         /* attr shouldn't be set on remote object */
 218         LASSERT(mdt_object_exists(mo) >= 0);
 219
 220         if (info->mti_epoch)
 221                 som_update = (info->mti_epoch->flags & MF_SOM_CHANGE);
 222
 223         /* Try to avoid object_lock if another epoch has been started
 224          * already. */
 225         if (som_update && (info->mti_epoch->ioepoch != mo->mot_ioepoch))
 226                 RETURN(0);
 227
 228         lh = &info->mti_lh[MDT_LH_PARENT];
 229         mdt_lock_reg_init(lh, LCK_PW);
 230
 231         if (!(flags & MRF_SETATTR_LOCKED)) {
 232                 __u64 lockpart = MDS_INODELOCK_UPDATE;
 233                 if (ma->ma_attr.la_valid & (LA_MODE|LA_UID|LA_GID))
 234                         lockpart |= MDS_INODELOCK_LOOKUP;
 235
 236                 rc = mdt_object_lock(info, mo, lh, lockpart, MDT_LOCAL_LOCK);
 237                 if (rc != 0)
 238                         RETURN(rc);
 239         }
 240
 241         /* Setattrs are syncronized through dlm lock taken above. If another
 242          * epoch started, its attributes may be already flushed on disk,
 243          * skip setattr. */
 244         if (som_update && (info->mti_epoch->ioepoch != mo->mot_ioepoch))
 245                 GOTO(out_unlock, rc = 0);
 246
 247         if (mdt_object_exists(mo) == 0)
 248                 GOTO(out_unlock, rc = -ENOENT);
 249
 250         /* all attrs are packed into mti_attr in unpack_setattr */
 251         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
 252                        OBD_FAIL_MDS_REINT_SETATTR_WRITE);
 253
 254         /* This is only for set ctime when rename's source is on remote MDS. */
 255         if (unlikely(ma->ma_attr.la_valid == LA_CTIME))
 256                 ma->ma_attr_flags |= MDS_VTX_BYPASS;
 257
 258         /* all attrs are packed into mti_attr in unpack_setattr */
 259         rc = mo_attr_set(info->mti_env, mdt_object_child(mo), ma);
 260         if (rc != 0)
 261                 GOTO(out_unlock, rc);
 262
 263         /* Re-enable SIZEONMDS. */
 264         if (som_update) {
 265                 CDEBUG(D_INODE, "Closing epoch "LPU64" on "DFID". Count %d\n",
 266                        mo->mot_ioepoch, PFID(mdt_object_fid(mo)),
 267                        mo->mot_epochcount);
 268                 mdt_sizeonmds_enable(info, mo);
 269         }
 270
 271         EXIT;
 272 out_unlock:
 273         mdt_object_unlock(info, mo, lh, rc);
 274         return rc;
 275 }
 276
 277 static int mdt_reint_setattr(struct mdt_thread_info *info,
 278                              struct mdt_lock_handle *lhc)
 279 {
 280         struct mdt_device       *mdt = info->mti_mdt;
 281         struct md_attr          *ma = &info->mti_attr;
 282         struct mdt_reint_record *rr = &info->mti_rr;
 283         struct ptlrpc_request   *req = mdt_info_req(info);
 284         struct mdt_export_data  *med = &req->rq_export->exp_mdt_data;
 285         struct mdt_file_data    *mfd;
 286         struct mdt_object       *mo;
 287         struct md_object        *next;
 288         struct mdt_body         *repbody;
 289         int                      rc;
 290         ENTRY;
 291
 292         DEBUG_REQ(D_INODE, req, "setattr "DFID" %x", PFID(rr->rr_fid1),
 293                   (unsigned int)ma->ma_attr.la_valid);
 294
 295         if (info->mti_dlm_req)
 296                 ldlm_request_cancel(req, info->mti_dlm_req, 0);
 297
 298         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
 299         mo = mdt_object_find(info->mti_env, info->mti_mdt, rr->rr_fid1);
 300         if (IS_ERR(mo))
 301                 GOTO(out, rc = PTR_ERR(mo));
 302
 303         if (!(mdt_conn_flags(info) & OBD_CONNECT_SOM)) {
 304                 if ((ma->ma_attr.la_valid & LA_SIZE) ||
 305                     (rr->rr_flags & MRF_SETATTR_LOCKED)) {
 306                         /* Check write access for the O_TRUNC case */
 307                         if (mdt_write_read(info->mti_mdt, mo) < 0)
 308                                 GOTO(out_put, rc = -ETXTBSY);
 309                 }
 310         } else if (info->mti_epoch &&
 311                     (info->mti_epoch->flags & MF_EPOCH_OPEN)) {
 312                 /* Truncate case. */
 313                 rc = mdt_write_get(info->mti_mdt, mo);
 314                 if (rc)
 315                         GOTO(out_put, rc);
 316
 317                 mfd = mdt_mfd_new();
 318                 if (mfd == NULL)
 319                         GOTO(out_put, rc = -ENOMEM);
 320
 321                 mdt_epoch_open(info, mo);
 322                 repbody->ioepoch = mo->mot_ioepoch;
 323
 324                 mdt_object_get(info->mti_env, mo);
 325                 mdt_mfd_set_mode(mfd, FMODE_EPOCHLCK);
 326                 mfd->mfd_object = mo;
 327                 mfd->mfd_xid = req->rq_xid;
 328
 329                 spin_lock(&med->med_open_lock);
 330                 list_add(&mfd->mfd_list, &med->med_open_head);
 331                 spin_unlock(&med->med_open_lock);
 332                 repbody->handle.cookie = mfd->mfd_handle.h_cookie;
 333         }
 334
 335         if (info->mti_epoch && (info->mti_epoch->flags & MF_SOM_CHANGE))
 336                 ma->ma_attr_flags |= MDS_PERM_BYPASS | MDS_SOM;
 337
 338         rc = mdt_attr_set(info, mo, rr->rr_flags);
 339         if (rc)
 340                 GOTO(out_put, rc);
 341
 342         if (info->mti_epoch && (info->mti_epoch->flags & MF_SOM_CHANGE)) {
 343                 LASSERT(mdt_conn_flags(info) & OBD_CONNECT_SOM);
 344                 LASSERT(info->mti_epoch);
 345
 346                 spin_lock(&med->med_open_lock);
 347                 /* Size-on-MDS Update. Find and free mfd. */
 348                 mfd = mdt_handle2mfd(info, &info->mti_epoch->handle);
 349                 if (mfd == NULL) {
 350                         spin_unlock(&med->med_open_lock);
 351                         CDEBUG(D_INODE | D_ERROR, "no handle for file close: "
 352                                         "fid = "DFID": cookie = "LPX64"\n",
 353                                         PFID(info->mti_rr.rr_fid1),
 354                                         info->mti_epoch->handle.cookie);
 355                         GOTO(out_put, rc = -ESTALE);
 356                 }
 357                 LASSERT(mfd->mfd_mode == FMODE_SOM);
 358                 LASSERT(!(info->mti_epoch->flags & MF_EPOCH_CLOSE));
 359
 360                 class_handle_unhash(&mfd->mfd_handle);
 361                 list_del_init(&mfd->mfd_list);
 362                 spin_unlock(&med->med_open_lock);
 363
 364                 mdt_mfd_close(info, mfd);
 365         }
 366
 367         ma->ma_need = MA_INODE;
 368         ma->ma_valid = 0;
 369         next = mdt_object_child(mo);
 370         rc = mo_attr_get(info->mti_env, next, ma);
 371         if (rc != 0)
 372                 GOTO(out_put, rc);
 373
 374         mdt_pack_attr2body(info, repbody, &ma->ma_attr, mdt_object_fid(mo));
 375
 376         if (mdt->mdt_opts.mo_oss_capa &&
 377             S_ISREG(lu_object_attr(&mo->mot_obj.mo_lu)) &&
 378             (ma->ma_attr.la_valid & LA_SIZE)) {
 379                 struct lustre_capa *capa;
 380
 381                 capa = req_capsule_server_get(info->mti_pill, &RMF_CAPA2);
 382                 LASSERT(capa);
 383                 capa->lc_opc = CAPA_OPC_OSS_DEFAULT | CAPA_OPC_OSS_TRUNC;
 384                 rc = mo_capa_get(info->mti_env, mdt_object_child(mo), capa, 0);
 385                 if (rc)
 386                         GOTO(out_put, rc);
 387                 repbody->valid |= OBD_MD_FLOSSCAPA;
 388         }
 389
 390         EXIT;
 391 out_put:
 392         mdt_object_put(info->mti_env, mo);
 393 out:
 394         mdt_shrink_reply(info);
 395         return rc;
 396 }
 397
 398 static int mdt_reint_create(struct mdt_thread_info *info,
 399                             struct mdt_lock_handle *lhc)
 400 {
 401         int rc;
 402         ENTRY;
 403
 404         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_CREATE))
 405                 RETURN(err_serious(-ESTALE));
 406
 407         if (info->mti_dlm_req)
 408                 ldlm_request_cancel(mdt_info_req(info), info->mti_dlm_req, 0);
 409
 410         switch (info->mti_attr.ma_attr.la_mode & S_IFMT) {
 411         case S_IFDIR:{
 412                 /* Cross-ref case. */
 413                 if (info->mti_cross_ref) {
 414                         rc = mdt_md_mkobj(info);
 415                         break;
 416                 }
 417         }
 418         case S_IFREG:
 419         case S_IFLNK:
 420         case S_IFCHR:
 421         case S_IFBLK:
 422         case S_IFIFO:
 423         case S_IFSOCK:{
 424                 /* Special file should stay on the same node as parent. */
 425                 LASSERT(info->mti_rr.rr_namelen > 0);
 426                 rc = mdt_md_create(info);
 427                 break;
 428         }
 429         default:
 430                 rc = err_serious(-EOPNOTSUPP);
 431         }
 432         RETURN(rc);
 433 }
 434
 435 static int mdt_reint_unlink(struct mdt_thread_info *info,
 436                             struct mdt_lock_handle *lhc)
 437 {
 438         struct mdt_reint_record *rr = &info->mti_rr;
 439         struct ptlrpc_request   *req = mdt_info_req(info);
 440         struct md_attr          *ma = &info->mti_attr;
 441         struct lu_fid           *child_fid = &info->mti_tmp_fid1;
 442         struct mdt_object       *mp;
 443         struct mdt_object       *mc;
 444         struct mdt_lock_handle  *parent_lh;
 445         struct mdt_lock_handle  *child_lh;
 446         struct lu_name          *lname;
 447         int                      rc;
 448         ENTRY;
 449
 450         DEBUG_REQ(D_INODE, req, "unlink "DFID"/%s", PFID(rr->rr_fid1),
 451                   rr->rr_name);
 452
 453         if (info->mti_dlm_req)
 454                 ldlm_request_cancel(req, info->mti_dlm_req, 0);
 455
 456         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNLINK))
 457                 RETURN(err_serious(-ENOENT));
 458
 459         /* step 1: lock the parent */
 460         parent_lh = &info->mti_lh[MDT_LH_PARENT];
 461         mdt_lock_pdo_init(parent_lh, LCK_PW, rr->rr_name,
 462                           rr->rr_namelen);
 463
 464         mp = mdt_object_find_lock(info, rr->rr_fid1, parent_lh,
 465                                   MDS_INODELOCK_UPDATE);
 466         if (IS_ERR(mp)) {
 467                 rc = PTR_ERR(mp);
 468                 /* errors are possible here in cross-ref cases, see below */
 469                 if (info->mti_cross_ref)
 470                         rc = 0;
 471                 GOTO(out, rc);
 472         }
 473
 474         mdt_reint_init_ma(info, ma);
 475         if (!ma->ma_lmm || !ma->ma_cookie)
 476                 GOTO(out_unlock_parent, rc = -EINVAL);
 477
 478         if (info->mti_cross_ref) {
 479                 /*
 480                  * Remote partial operation. It is possible that replay may
 481                  * happen on parent MDT and this operation will be repeated.
 482                  * Therefore the object absense is allowed case and nothing
 483                  * should be done here.
 484                  */
 485                 if (mdt_object_exists(mp) > 0) {
 486                         mdt_set_capainfo(info, 0, rr->rr_fid1, BYPASS_CAPA);
 487                         rc = mo_ref_del(info->mti_env,
 488                                         mdt_object_child(mp), ma);
 489                         mdt_handle_last_unlink(info, mp, ma);
 490                 } else
 491                         rc = 0;
 492                 GOTO(out_unlock_parent, rc);
 493         }
 494
 495         /* step 2: find & lock the child */
 496         lname = mdt_name(info->mti_env, (char *)rr->rr_name, rr->rr_namelen);
 497         rc = mdo_lookup(info->mti_env, mdt_object_child(mp),
 498                         lname, child_fid, &info->mti_spec);
 499         if (rc != 0)
 500                  GOTO(out_unlock_parent, rc);
 501
 502         /* We will lock the child regardless it is local or remote. No harm. */
 503         mc = mdt_object_find(info->mti_env, info->mti_mdt, child_fid);
 504         if (IS_ERR(mc))
 505                 GOTO(out_unlock_parent, rc = PTR_ERR(mc));
 506         child_lh = &info->mti_lh[MDT_LH_CHILD];
 507         mdt_lock_reg_init(child_lh, LCK_EX);
 508         rc = mdt_object_lock(info, mc, child_lh, MDS_INODELOCK_FULL,
 509                              MDT_CROSS_LOCK);
 510         if (rc != 0) {
 511                 mdt_object_put(info->mti_env, mc);
 512                 GOTO(out_unlock_parent, rc);
 513         }
 514
 515         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
 516                        OBD_FAIL_MDS_REINT_UNLINK_WRITE);
 517
 518         /*
 519          * Now we can only make sure we need MA_INODE, in mdd layer, will check
 520          * whether need MA_LOV and MA_COOKIE.
 521          */
 522         ma->ma_need = MA_INODE;
 523         ma->ma_valid = 0;
 524         mdt_set_capainfo(info, 1, child_fid, BYPASS_CAPA);
 525         rc = mdo_unlink(info->mti_env, mdt_object_child(mp),
 526                         mdt_object_child(mc), lname, ma);
 527         if (rc == 0)
 528                 mdt_handle_last_unlink(info, mc, ma);
 529
 530         EXIT;
 531         mdt_object_unlock_put(info, mc, child_lh, rc);
 532 out_unlock_parent:
 533         mdt_object_unlock_put(info, mp, parent_lh, rc);
 534 out:
 535         return rc;
 536 }
 537
 538 static int mdt_reint_link(struct mdt_thread_info *info,
 539                           struct mdt_lock_handle *lhc)
 540 {
 541         struct mdt_reint_record *rr = &info->mti_rr;
 542         struct ptlrpc_request   *req = mdt_info_req(info);
 543         struct md_attr          *ma = &info->mti_attr;
 544         struct mdt_object       *ms;
 545         struct mdt_object       *mp;
 546         struct mdt_lock_handle  *lhs;
 547         struct mdt_lock_handle  *lhp;
 548         struct lu_name          *lname;
 549         int rc;
 550         ENTRY;
 551
 552         DEBUG_REQ(D_INODE, req, "link "DFID" to "DFID"/%s",
 553                   PFID(rr->rr_fid1), PFID(rr->rr_fid2), rr->rr_name);
 554
 555         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_LINK))
 556                 RETURN(err_serious(-ENOENT));
 557
 558         if (info->mti_dlm_req)
 559                 ldlm_request_cancel(req, info->mti_dlm_req, 0);
 560
 561         if (info->mti_cross_ref) {
 562                 /* MDT holding name ask us to add ref. */
 563                 lhs = &info->mti_lh[MDT_LH_CHILD];
 564                 mdt_lock_reg_init(lhs, LCK_EX);
 565                 ms = mdt_object_find_lock(info, rr->rr_fid1, lhs,
 566                                           MDS_INODELOCK_UPDATE);
 567                 if (IS_ERR(ms))
 568                         RETURN(PTR_ERR(ms));
 569
 570                 mdt_set_capainfo(info, 0, rr->rr_fid1, BYPASS_CAPA);
 571                 rc = mo_ref_add(info->mti_env, mdt_object_child(ms), ma);
 572                 mdt_object_unlock_put(info, ms, lhs, rc);
 573                 RETURN(rc);
 574         }
 575
 576         /* Invalid case so return error immediately instead of
 577          * processing it */
 578         if (lu_fid_eq(rr->rr_fid1, rr->rr_fid2))
 579                 RETURN(-EPERM);
 580
 581         /* step 1: find & lock the target parent dir */
 582         lhp = &info->mti_lh[MDT_LH_PARENT];
 583         mdt_lock_pdo_init(lhp, LCK_EX, rr->rr_name,
 584                           rr->rr_namelen);
 585         mp = mdt_object_find_lock(info, rr->rr_fid2, lhp,
 586                                   MDS_INODELOCK_UPDATE);
 587         if (IS_ERR(mp))
 588                 RETURN(PTR_ERR(mp));
 589
 590         /* step 2: find & lock the source */
 591         lhs = &info->mti_lh[MDT_LH_CHILD];
 592         mdt_lock_reg_init(lhs, LCK_EX);
 593
 594         ms = mdt_object_find(info->mti_env, info->mti_mdt, rr->rr_fid1);
 595         if (IS_ERR(ms))
 596                 GOTO(out_unlock_parent, rc = PTR_ERR(ms));
 597
 598         rc = mdt_object_lock(info, ms, lhs, MDS_INODELOCK_UPDATE,
 599                             MDT_CROSS_LOCK);
 600         if (rc != 0) {
 601                 mdt_object_put(info->mti_env, ms);
 602                 GOTO(out_unlock_parent, rc);
 603         }
 604
 605         /* step 3: link it */
 606         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
 607                        OBD_FAIL_MDS_REINT_LINK_WRITE);
 608
 609         lname = mdt_name(info->mti_env, (char *)rr->rr_name, rr->rr_namelen);
 610         rc = mdo_link(info->mti_env, mdt_object_child(mp),
 611                       mdt_object_child(ms), lname, ma);
 612
 613         EXIT;
 614         mdt_object_unlock_put(info, ms, lhs, rc);
 615 out_unlock_parent:
 616         mdt_object_unlock_put(info, mp, lhp, rc);
 617         return rc;
 618 }
 619
 620 /* partial operation for rename */
 621 static int mdt_reint_rename_tgt(struct mdt_thread_info *info)
 622 {
 623         struct mdt_reint_record *rr = &info->mti_rr;
 624         struct ptlrpc_request   *req = mdt_info_req(info);
 625         struct md_attr          *ma = &info->mti_attr;
 626         struct mdt_object       *mtgtdir;
 627         struct mdt_object       *mtgt = NULL;
 628         struct mdt_lock_handle  *lh_tgtdir;
 629         struct mdt_lock_handle  *lh_tgt = NULL;
 630         struct lu_fid           *tgt_fid = &info->mti_tmp_fid1;
 631         struct lu_name          *lname;
 632         int                      rc;
 633         ENTRY;
 634
 635         DEBUG_REQ(D_INODE, req, "rename_tgt: insert (%s->"DFID") in "DFID,
 636                   rr->rr_tgt, PFID(rr->rr_fid2), PFID(rr->rr_fid1));
 637
 638         /* step 1: lookup & lock the tgt dir. */
 639         lh_tgtdir = &info->mti_lh[MDT_LH_PARENT];
 640         mdt_lock_pdo_init(lh_tgtdir, LCK_PW, rr->rr_tgt,
 641                           rr->rr_tgtlen);
 642         mtgtdir = mdt_object_find_lock(info, rr->rr_fid1, lh_tgtdir,
 643                                        MDS_INODELOCK_UPDATE);
 644         if (IS_ERR(mtgtdir))
 645                 RETURN(PTR_ERR(mtgtdir));
 646
 647         /* step 2: find & lock the target object if exists. */
 648         mdt_set_capainfo(info, 0, rr->rr_fid1, BYPASS_CAPA);
 649         lname = mdt_name(info->mti_env, (char *)rr->rr_tgt, rr->rr_tgtlen);
 650         rc = mdo_lookup(info->mti_env, mdt_object_child(mtgtdir),
 651                         lname, tgt_fid, &info->mti_spec);
 652         if (rc != 0 && rc != -ENOENT) {
 653                 GOTO(out_unlock_tgtdir, rc);
 654         } else if (rc == 0) {
 655                 /*
 656                  * In case of replay that name can be already inserted, check
 657                  * that and do nothing if so.
 658                  */
 659                 if (lu_fid_eq(tgt_fid, rr->rr_fid2))
 660                         GOTO(out_unlock_tgtdir, rc);
 661
 662                 lh_tgt = &info->mti_lh[MDT_LH_CHILD];
 663                 mdt_lock_reg_init(lh_tgt, LCK_EX);
 664
 665                 mtgt = mdt_object_find_lock(info, tgt_fid, lh_tgt,
 666                                             MDS_INODELOCK_LOOKUP);
 667                 if (IS_ERR(mtgt))
 668                         GOTO(out_unlock_tgtdir, rc = PTR_ERR(mtgt));
 669
 670                 mdt_reint_init_ma(info, ma);
 671                 if (!ma->ma_lmm || !ma->ma_cookie)
 672                         GOTO(out_unlock_tgt, rc = -EINVAL);
 673
 674                 rc = mdo_rename_tgt(info->mti_env, mdt_object_child(mtgtdir),
 675                                     mdt_object_child(mtgt), rr->rr_fid2,
 676                                     lname, ma);
 677         } else /* -ENOENT */ {
 678                 rc = mdo_name_insert(info->mti_env, mdt_object_child(mtgtdir),
 679                                      lname, rr->rr_fid2, ma);
 680         }
 681
 682         /* handle last link of tgt object */
 683         if (rc == 0 && mtgt)
 684                 mdt_handle_last_unlink(info, mtgt, ma);
 685
 686         EXIT;
 687 out_unlock_tgt:
 688         if (mtgt)
 689                 mdt_object_unlock_put(info, mtgt, lh_tgt, rc);
 690 out_unlock_tgtdir:
 691         mdt_object_unlock_put(info, mtgtdir, lh_tgtdir, rc);
 692         return rc;
 693 }
 694
 695 static int mdt_rename_lock(struct mdt_thread_info *info,
 696                            struct lustre_handle *lh)
 697 {
 698         struct ldlm_namespace *ns     = info->mti_mdt->mdt_namespace;
 699         ldlm_policy_data_t    *policy = &info->mti_policy;
 700         struct ldlm_res_id    *res_id = &info->mti_res_id;
 701         struct lu_site        *ls;
 702         int rc;
 703         ENTRY;
 704
 705         /*
 706          * Disable global rename BFL lock temporarily because
 707          * when a mds do rename recoverying, which might enqueue
 708          * BFL lock to the controller mds. and this req might be
 709          * replay req for controller mds. but we did not have
 710          * such handling in controller mds. XXX
 711          */
 712         RETURN(0);
 713         ls = info->mti_mdt->mdt_md_dev.md_lu_dev.ld_site;
 714         fid_build_reg_res_name(&LUSTRE_BFL_FID, res_id);
 715
 716         memset(policy, 0, sizeof *policy);
 717         policy->l_inodebits.bits = MDS_INODELOCK_UPDATE;
 718
 719         if (ls->ls_control_exp == NULL) {
 720                 int flags = LDLM_FL_LOCAL_ONLY | LDLM_FL_ATOMIC_CB;
 721
 722                 /*
 723                  * Current node is controller, that is mdt0, where we should
 724                  * take BFL lock.
 725                  */
 726                 rc = ldlm_cli_enqueue_local(ns, res_id, LDLM_IBITS, policy,
 727                                             LCK_EX, &flags, ldlm_blocking_ast,
 728                                             ldlm_completion_ast, NULL, NULL, 0,
 729                                             NULL, lh);
 730         } else {
 731                 struct ldlm_enqueue_info einfo = { LDLM_IBITS, LCK_EX,
 732                         ldlm_blocking_ast, ldlm_completion_ast, NULL, NULL };
 733                 int flags = 0;
 734
 735                 /*
 736                  * This is the case mdt0 is remote node, issue DLM lock like
 737                  * other clients.
 738                  */
 739                 rc = ldlm_cli_enqueue(ls->ls_control_exp, NULL, &einfo, res_id,
 740                                       policy, &flags, NULL, 0, NULL, lh, 0);
 741         }
 742
 743         RETURN(rc);
 744 }
 745
 746 static void mdt_rename_unlock(struct lustre_handle *lh)
 747 {
 748         ENTRY;
 749         /* Disable global rename BFL lock temporarily. see above XXX*/
 750         EXIT;
 751         return;
 752         LASSERT(lustre_handle_is_used(lh));
 753         ldlm_lock_decref(lh, LCK_EX);
 754         EXIT;
 755 }
 756
 757 /*
 758  * This is is_subdir() variant, it is CMD if cmm forwards it to correct
 759  * target. Source should not be ancestor of target dir. May be other rename
 760  * checks can be moved here later.
 761  */
 762 static int mdt_rename_sanity(struct mdt_thread_info *info, struct lu_fid *fid)
 763 {
 764         struct mdt_reint_record *rr = &info->mti_rr;
 765         struct lu_fid dst_fid = *rr->rr_fid2;
 766         struct mdt_object *dst;
 767         int rc = 0;
 768         ENTRY;
 769
 770         do {
 771                 LASSERT(fid_is_sane(&dst_fid));
 772                 dst = mdt_object_find(info->mti_env, info->mti_mdt, &dst_fid);
 773                 if (!IS_ERR(dst)) {
 774                         rc = mdo_is_subdir(info->mti_env,
 775                                            mdt_object_child(dst), fid,
 776                                            &dst_fid);
 777                         mdt_object_put(info->mti_env, dst);
 778                         if (rc != -EREMOTE && rc < 0) {
 779                                 CERROR("Failed mdo_is_subdir(), rc %d\n", rc);
 780                         } else {
 781                                 /* check the found fid */
 782                                 if (lu_fid_eq(&dst_fid, fid))
 783                                         rc = -EINVAL;
 784                         }
 785                 } else {
 786                         rc = PTR_ERR(dst);
 787                 }
 788         } while (rc == -EREMOTE);
 789
 790         RETURN(rc);
 791 }
 792
 793 static int mdt_reint_rename(struct mdt_thread_info *info,
 794                             struct mdt_lock_handle *lhc)
 795 {
 796         struct mdt_reint_record *rr = &info->mti_rr;
 797         struct md_attr          *ma = &info->mti_attr;
 798         struct ptlrpc_request   *req = mdt_info_req(info);
 799         struct mdt_object       *msrcdir;
 800         struct mdt_object       *mtgtdir;
 801         struct mdt_object       *mold;
 802         struct mdt_object       *mnew = NULL;
 803         struct mdt_lock_handle  *lh_srcdirp;
 804         struct mdt_lock_handle  *lh_tgtdirp;
 805         struct mdt_lock_handle  *lh_oldp;
 806         struct mdt_lock_handle  *lh_newp;
 807         struct lu_fid           *old_fid = &info->mti_tmp_fid1;
 808         struct lu_fid           *new_fid = &info->mti_tmp_fid2;
 809         struct lustre_handle     rename_lh = { 0 };
 810         struct lu_name           slname = { 0 };
 811         struct lu_name          *lname;
 812         int                      rc;
 813         ENTRY;
 814
 815         if (info->mti_dlm_req)
 816                 ldlm_request_cancel(mdt_info_req(info), info->mti_dlm_req, 0);
 817
 818         if (info->mti_cross_ref) {
 819                 rc = mdt_reint_rename_tgt(info);
 820                 RETURN(rc);
 821         }
 822
 823         DEBUG_REQ(D_INODE, req, "rename "DFID"/%s to "DFID"/%s",
 824                   PFID(rr->rr_fid1), rr->rr_name,
 825                   PFID(rr->rr_fid2), rr->rr_tgt);
 826
 827         rc = mdt_rename_lock(info, &rename_lh);
 828         if (rc) {
 829                 CERROR("Can't lock FS for rename, rc %d\n", rc);
 830                 RETURN(rc);
 831         }
 832
 833         lh_newp = &info->mti_lh[MDT_LH_NEW];
 834
 835         /* step 1: lock the source dir. */
 836         lh_srcdirp = &info->mti_lh[MDT_LH_PARENT];
 837         mdt_lock_pdo_init(lh_srcdirp, LCK_PW, rr->rr_name,
 838                           rr->rr_namelen);
 839         msrcdir = mdt_object_find_lock(info, rr->rr_fid1, lh_srcdirp,
 840                                        MDS_INODELOCK_UPDATE);
 841         if (IS_ERR(msrcdir))
 842                 GOTO(out_rename_lock, rc = PTR_ERR(msrcdir));
 843
 844         /* step 2: find & lock the target dir. */
 845         lh_tgtdirp = &info->mti_lh[MDT_LH_CHILD];
 846         mdt_lock_pdo_init(lh_tgtdirp, LCK_PW, rr->rr_tgt,
 847                           rr->rr_tgtlen);
 848         if (lu_fid_eq(rr->rr_fid1, rr->rr_fid2)) {
 849                 mdt_object_get(info->mti_env, msrcdir);
 850                 mtgtdir = msrcdir;
 851         } else {
 852                 mtgtdir = mdt_object_find(info->mti_env, info->mti_mdt,
 853                                           rr->rr_fid2);
 854                 if (IS_ERR(mtgtdir))
 855                         GOTO(out_unlock_source, rc = PTR_ERR(mtgtdir));
 856
 857                 rc = mdt_object_exists(mtgtdir);
 858                 if (rc == 0)
 859                         GOTO(out_unlock_target, rc = -ESTALE);
 860                 else if (rc > 0) {
 861                         /* we lock the target dir if it is local */
 862                         rc = mdt_object_lock(info, mtgtdir, lh_tgtdirp,
 863                                              MDS_INODELOCK_UPDATE,
 864                                              MDT_LOCAL_LOCK);
 865                         if (rc != 0)
 866                                 GOTO(out_unlock_target, rc);
 867                 }
 868         }
 869
 870         /* step 3: find & lock the old object. */
 871         lname = mdt_name(info->mti_env, (char *)rr->rr_name, rr->rr_namelen);
 872         mdt_name_copy(&slname, lname);
 873         rc = mdo_lookup(info->mti_env, mdt_object_child(msrcdir),
 874                         &slname, old_fid, &info->mti_spec);
 875         if (rc != 0)
 876                 GOTO(out_unlock_target, rc);
 877
 878         if (lu_fid_eq(old_fid, rr->rr_fid1) || lu_fid_eq(old_fid, rr->rr_fid2))
 879                 GOTO(out_unlock_target, rc = -EINVAL);
 880
 881         mold = mdt_object_find(info->mti_env, info->mti_mdt, old_fid);
 882         if (IS_ERR(mold))
 883                 GOTO(out_unlock_target, rc = PTR_ERR(mold));
 884
 885         lh_oldp = &info->mti_lh[MDT_LH_OLD];
 886         mdt_lock_reg_init(lh_oldp, LCK_EX);
 887         rc = mdt_object_lock(info, mold, lh_oldp, MDS_INODELOCK_LOOKUP,
 888                              MDT_CROSS_LOCK);
 889         if (rc != 0) {
 890                 mdt_object_put(info->mti_env, mold);
 891                 GOTO(out_unlock_target, rc);
 892         }
 893         mdt_set_capainfo(info, 2, old_fid, BYPASS_CAPA);
 894
 895         /* step 4: find & lock the new object. */
 896         /* new target object may not exist now */
 897         lname = mdt_name(info->mti_env, (char *)rr->rr_tgt, rr->rr_tgtlen);
 898         rc = mdo_lookup(info->mti_env, mdt_object_child(mtgtdir),
 899                         lname, new_fid, &info->mti_spec);
 900         if (rc == 0) {
 901                 /* the new_fid should have been filled at this moment */
 902                 if (lu_fid_eq(old_fid, new_fid))
 903                        GOTO(out_unlock_old, rc);
 904
 905                 if (lu_fid_eq(new_fid, rr->rr_fid1) ||
 906                     lu_fid_eq(new_fid, rr->rr_fid2))
 907                         GOTO(out_unlock_old, rc = -EINVAL);
 908
 909                 mdt_lock_reg_init(lh_newp, LCK_EX);
 910                 mnew = mdt_object_find(info->mti_env, info->mti_mdt, new_fid);
 911                 if (IS_ERR(mnew))
 912                         GOTO(out_unlock_old, rc = PTR_ERR(mnew));
 913
 914                 rc = mdt_object_lock(info, mnew, lh_newp,
 915                                      MDS_INODELOCK_FULL, MDT_CROSS_LOCK);
 916                 if (rc != 0) {
 917                         mdt_object_put(info->mti_env, mnew);
 918                         GOTO(out_unlock_old, rc);
 919                 }
 920                 mdt_set_capainfo(info, 3, new_fid, BYPASS_CAPA);
 921         } else if (rc != -EREMOTE && rc != -ENOENT)
 922                 GOTO(out_unlock_old, rc);
 923
 924         /* step 5: rename it */
 925         mdt_reint_init_ma(info, ma);
 926         if (!ma->ma_lmm || !ma->ma_cookie)
 927                 GOTO(out_unlock_new, rc = -EINVAL);
 928
 929         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
 930                        OBD_FAIL_MDS_REINT_RENAME_WRITE);
 931
 932
 933         /* Check if @dst is subdir of @src. */
 934         rc = mdt_rename_sanity(info, old_fid);
 935         if (rc)
 936                 GOTO(out_unlock_new, rc);
 937
 938         rc = mdo_rename(info->mti_env, mdt_object_child(msrcdir),
 939                         mdt_object_child(mtgtdir), old_fid, &slname,
 940                         (mnew ? mdt_object_child(mnew) : NULL),
 941                         lname, ma);
 942
 943         /* handle last link of tgt object */
 944         if (rc == 0 && mnew)
 945                 mdt_handle_last_unlink(info, mnew, ma);
 946
 947         EXIT;
 948 out_unlock_new:
 949         if (mnew)
 950                 mdt_object_unlock_put(info, mnew, lh_newp, rc);
 951 out_unlock_old:
 952         mdt_object_unlock_put(info, mold, lh_oldp, rc);
 953 out_unlock_target:
 954         mdt_object_unlock_put(info, mtgtdir, lh_tgtdirp, rc);
 955 out_unlock_source:
 956         mdt_object_unlock_put(info, msrcdir, lh_srcdirp, rc);
 957 out_rename_lock:
 958         mdt_rename_unlock(&rename_lh);
 959         return rc;
 960 }
 961
 962 typedef int (*mdt_reinter)(struct mdt_thread_info *info,
 963                            struct mdt_lock_handle *lhc);
 964
 965 static mdt_reinter reinters[REINT_MAX] = {
 966         [REINT_SETATTR]  = mdt_reint_setattr,
 967         [REINT_CREATE]   = mdt_reint_create,
 968         [REINT_LINK]     = mdt_reint_link,
 969         [REINT_UNLINK]   = mdt_reint_unlink,
 970         [REINT_RENAME]   = mdt_reint_rename,
 971         [REINT_OPEN]     = mdt_reint_open,
 972         [REINT_SETXATTR] = mdt_reint_setxattr
 973 };
 974
 975 int mdt_reint_rec(struct mdt_thread_info *info,
 976                   struct mdt_lock_handle *lhc)
 977 {
 978         int rc;
 979         ENTRY;
 980
 981         rc = reinters[info->mti_rr.rr_opcode](info, lhc);
 982
 983         RETURN(rc);
 984 }