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         rc = mdt_attr_set(info, mo, rr->rr_flags);
 336         if (rc)
 337                 GOTO(out_put, rc);
 338
 339         if (info->mti_epoch && (info->mti_epoch->flags & MF_SOM_CHANGE)) {
 340                 LASSERT(mdt_conn_flags(info) & OBD_CONNECT_SOM);
 341                 LASSERT(info->mti_epoch);
 342
 343                 spin_lock(&med->med_open_lock);
 344                 /* Size-on-MDS Update. Find and free mfd. */
 345                 mfd = mdt_handle2mfd(info, &info->mti_epoch->handle);
 346                 if (mfd == NULL) {
 347                         spin_unlock(&med->med_open_lock);
 348                         CDEBUG(D_INODE | D_ERROR, "no handle for file close: "
 349                                         "fid = "DFID": cookie = "LPX64"\n",
 350                                         PFID(info->mti_rr.rr_fid1),
 351                                         info->mti_epoch->handle.cookie);
 352                         GOTO(out_put, rc = -ESTALE);
 353                 }
 354                 LASSERT(mfd->mfd_mode == FMODE_SOM);
 355                 LASSERT(!(info->mti_epoch->flags & MF_EPOCH_CLOSE));
 356
 357                 class_handle_unhash(&mfd->mfd_handle);
 358                 list_del_init(&mfd->mfd_list);
 359                 spin_unlock(&med->med_open_lock);
 360
 361                 mdt_mfd_close(info, mfd);
 362         }
 363
 364         ma->ma_need = MA_INODE;
 365         ma->ma_valid = 0;
 366         next = mdt_object_child(mo);
 367         rc = mo_attr_get(info->mti_env, next, ma);
 368         if (rc != 0)
 369                 GOTO(out_put, rc);
 370
 371         mdt_pack_attr2body(info, repbody, &ma->ma_attr, mdt_object_fid(mo));
 372
 373         if (mdt->mdt_opts.mo_oss_capa &&
 374             S_ISREG(lu_object_attr(&mo->mot_obj.mo_lu)) &&
 375             (ma->ma_attr.la_valid & LA_SIZE)) {
 376                 struct lustre_capa *capa;
 377
 378                 capa = req_capsule_server_get(info->mti_pill, &RMF_CAPA2);
 379                 LASSERT(capa);
 380                 capa->lc_opc = CAPA_OPC_OSS_DEFAULT | CAPA_OPC_OSS_TRUNC;
 381                 rc = mo_capa_get(info->mti_env, mdt_object_child(mo), capa, 0);
 382                 if (rc)
 383                         GOTO(out_put, rc);
 384                 repbody->valid |= OBD_MD_FLOSSCAPA;
 385         }
 386
 387         EXIT;
 388 out_put:
 389         mdt_object_put(info->mti_env, mo);
 390 out:
 391         mdt_shrink_reply(info);
 392         return rc;
 393 }
 394
 395 static int mdt_reint_create(struct mdt_thread_info *info,
 396                             struct mdt_lock_handle *lhc)
 397 {
 398         int rc;
 399         ENTRY;
 400
 401         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_CREATE))
 402                 RETURN(err_serious(-ESTALE));
 403
 404         if (info->mti_dlm_req)
 405                 ldlm_request_cancel(mdt_info_req(info), info->mti_dlm_req, 0);
 406
 407         switch (info->mti_attr.ma_attr.la_mode & S_IFMT) {
 408         case S_IFDIR:{
 409                 /* Cross-ref case. */
 410                 if (info->mti_cross_ref) {
 411                         rc = mdt_md_mkobj(info);
 412                         break;
 413                 }
 414         }
 415         case S_IFREG:
 416         case S_IFLNK:
 417         case S_IFCHR:
 418         case S_IFBLK:
 419         case S_IFIFO:
 420         case S_IFSOCK:{
 421                 /* Special file should stay on the same node as parent. */
 422                 LASSERT(info->mti_rr.rr_namelen > 0);
 423                 rc = mdt_md_create(info);
 424                 break;
 425         }
 426         default:
 427                 rc = err_serious(-EOPNOTSUPP);
 428         }
 429         RETURN(rc);
 430 }
 431
 432 static int mdt_reint_unlink(struct mdt_thread_info *info,
 433                             struct mdt_lock_handle *lhc)
 434 {
 435         struct mdt_reint_record *rr = &info->mti_rr;
 436         struct ptlrpc_request   *req = mdt_info_req(info);
 437         struct md_attr          *ma = &info->mti_attr;
 438         struct lu_fid           *child_fid = &info->mti_tmp_fid1;
 439         struct mdt_object       *mp;
 440         struct mdt_object       *mc;
 441         struct mdt_lock_handle  *parent_lh;
 442         struct mdt_lock_handle  *child_lh;
 443         struct lu_name          *lname;
 444         int                      rc;
 445         ENTRY;
 446
 447         DEBUG_REQ(D_INODE, req, "unlink "DFID"/%s", PFID(rr->rr_fid1),
 448                   rr->rr_name);
 449
 450         if (info->mti_dlm_req)
 451                 ldlm_request_cancel(req, info->mti_dlm_req, 0);
 452
 453         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNLINK))
 454                 RETURN(err_serious(-ENOENT));
 455
 456         /* step 1: lock the parent */
 457         parent_lh = &info->mti_lh[MDT_LH_PARENT];
 458         mdt_lock_pdo_init(parent_lh, LCK_PW, rr->rr_name,
 459                           rr->rr_namelen);
 460
 461         mp = mdt_object_find_lock(info, rr->rr_fid1, parent_lh,
 462                                   MDS_INODELOCK_UPDATE);
 463         if (IS_ERR(mp)) {
 464                 rc = PTR_ERR(mp);
 465                 /* errors are possible here in cross-ref cases, see below */
 466                 if (info->mti_cross_ref)
 467                         rc = 0;
 468                 GOTO(out, rc);
 469         }
 470
 471         mdt_reint_init_ma(info, ma);
 472         if (!ma->ma_lmm || !ma->ma_cookie)
 473                 GOTO(out_unlock_parent, rc = -EINVAL);
 474
 475         if (info->mti_cross_ref) {
 476                 /*
 477                  * Remote partial operation. It is possible that replay may
 478                  * happen on parent MDT and this operation will be repeated.
 479                  * Therefore the object absense is allowed case and nothing
 480                  * should be done here.
 481                  */
 482                 if (mdt_object_exists(mp) > 0) {
 483                         mdt_set_capainfo(info, 0, rr->rr_fid1, BYPASS_CAPA);
 484                         rc = mo_ref_del(info->mti_env,
 485                                         mdt_object_child(mp), ma);
 486                         mdt_handle_last_unlink(info, mp, ma);
 487                 } else
 488                         rc = 0;
 489                 GOTO(out_unlock_parent, rc);
 490         }
 491
 492         /* step 2: find & lock the child */
 493         lname = mdt_name(info->mti_env, (char *)rr->rr_name, rr->rr_namelen);
 494         rc = mdo_lookup(info->mti_env, mdt_object_child(mp),
 495                         lname, child_fid, &info->mti_spec);
 496         if (rc != 0)
 497                  GOTO(out_unlock_parent, rc);
 498
 499         /* We will lock the child regardless it is local or remote. No harm. */
 500         mc = mdt_object_find(info->mti_env, info->mti_mdt, child_fid);
 501         if (IS_ERR(mc))
 502                 GOTO(out_unlock_parent, rc = PTR_ERR(mc));
 503         child_lh = &info->mti_lh[MDT_LH_CHILD];
 504         mdt_lock_reg_init(child_lh, LCK_EX);
 505         rc = mdt_object_lock(info, mc, child_lh, MDS_INODELOCK_FULL,
 506                              MDT_CROSS_LOCK);
 507         if (rc != 0) {
 508                 mdt_object_put(info->mti_env, mc);
 509                 GOTO(out_unlock_parent, rc);
 510         }
 511
 512         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
 513                        OBD_FAIL_MDS_REINT_UNLINK_WRITE);
 514
 515         /*
 516          * Now we can only make sure we need MA_INODE, in mdd layer, will check
 517          * whether need MA_LOV and MA_COOKIE.
 518          */
 519         ma->ma_need = MA_INODE;
 520         ma->ma_valid = 0;
 521         mdt_set_capainfo(info, 1, child_fid, BYPASS_CAPA);
 522         rc = mdo_unlink(info->mti_env, mdt_object_child(mp),
 523                         mdt_object_child(mc), lname, ma);
 524         if (rc == 0)
 525                 mdt_handle_last_unlink(info, mc, ma);
 526
 527         EXIT;
 528         mdt_object_unlock_put(info, mc, child_lh, rc);
 529 out_unlock_parent:
 530         mdt_object_unlock_put(info, mp, parent_lh, rc);
 531 out:
 532         return rc;
 533 }
 534
 535 static int mdt_reint_link(struct mdt_thread_info *info,
 536                           struct mdt_lock_handle *lhc)
 537 {
 538         struct mdt_reint_record *rr = &info->mti_rr;
 539         struct ptlrpc_request   *req = mdt_info_req(info);
 540         struct md_attr          *ma = &info->mti_attr;
 541         struct mdt_object       *ms;
 542         struct mdt_object       *mp;
 543         struct mdt_lock_handle  *lhs;
 544         struct mdt_lock_handle  *lhp;
 545         struct lu_name          *lname;
 546         int rc;
 547         ENTRY;
 548
 549         DEBUG_REQ(D_INODE, req, "link "DFID" to "DFID"/%s",
 550                   PFID(rr->rr_fid1), PFID(rr->rr_fid2), rr->rr_name);
 551
 552         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_LINK))
 553                 RETURN(err_serious(-ENOENT));
 554
 555         if (info->mti_dlm_req)
 556                 ldlm_request_cancel(req, info->mti_dlm_req, 0);
 557
 558         if (info->mti_cross_ref) {
 559                 /* MDT holding name ask us to add ref. */
 560                 lhs = &info->mti_lh[MDT_LH_CHILD];
 561                 mdt_lock_reg_init(lhs, LCK_EX);
 562                 ms = mdt_object_find_lock(info, rr->rr_fid1, lhs,
 563                                           MDS_INODELOCK_UPDATE);
 564                 if (IS_ERR(ms))
 565                         RETURN(PTR_ERR(ms));
 566
 567                 mdt_set_capainfo(info, 0, rr->rr_fid1, BYPASS_CAPA);
 568                 rc = mo_ref_add(info->mti_env, mdt_object_child(ms), ma);
 569                 mdt_object_unlock_put(info, ms, lhs, rc);
 570                 RETURN(rc);
 571         }
 572
 573         /* Invalid case so return error immediately instead of
 574          * processing it */
 575         if (lu_fid_eq(rr->rr_fid1, rr->rr_fid2))
 576                 RETURN(-EPERM);
 577
 578         /* step 1: find & lock the target parent dir */
 579         lhp = &info->mti_lh[MDT_LH_PARENT];
 580         mdt_lock_pdo_init(lhp, LCK_EX, rr->rr_name,
 581                           rr->rr_namelen);
 582         mp = mdt_object_find_lock(info, rr->rr_fid2, lhp,
 583                                   MDS_INODELOCK_UPDATE);
 584         if (IS_ERR(mp))
 585                 RETURN(PTR_ERR(mp));
 586
 587         /* step 2: find & lock the source */
 588         lhs = &info->mti_lh[MDT_LH_CHILD];
 589         mdt_lock_reg_init(lhs, LCK_EX);
 590
 591         ms = mdt_object_find(info->mti_env, info->mti_mdt, rr->rr_fid1);
 592         if (IS_ERR(ms))
 593                 GOTO(out_unlock_parent, rc = PTR_ERR(ms));
 594
 595         rc = mdt_object_lock(info, ms, lhs, MDS_INODELOCK_UPDATE,
 596                             MDT_CROSS_LOCK);
 597         if (rc != 0) {
 598                 mdt_object_put(info->mti_env, ms);
 599                 GOTO(out_unlock_parent, rc);
 600         }
 601
 602         /* step 3: link it */
 603         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
 604                        OBD_FAIL_MDS_REINT_LINK_WRITE);
 605
 606         lname = mdt_name(info->mti_env, (char *)rr->rr_name, rr->rr_namelen);
 607         rc = mdo_link(info->mti_env, mdt_object_child(mp),
 608                       mdt_object_child(ms), lname, ma);
 609
 610         EXIT;
 611         mdt_object_unlock_put(info, ms, lhs, rc);
 612 out_unlock_parent:
 613         mdt_object_unlock_put(info, mp, lhp, rc);
 614         return rc;
 615 }
 616
 617 /* partial operation for rename */
 618 static int mdt_reint_rename_tgt(struct mdt_thread_info *info)
 619 {
 620         struct mdt_reint_record *rr = &info->mti_rr;
 621         struct ptlrpc_request   *req = mdt_info_req(info);
 622         struct md_attr          *ma = &info->mti_attr;
 623         struct mdt_object       *mtgtdir;
 624         struct mdt_object       *mtgt = NULL;
 625         struct mdt_lock_handle  *lh_tgtdir;
 626         struct mdt_lock_handle  *lh_tgt = NULL;
 627         struct lu_fid           *tgt_fid = &info->mti_tmp_fid1;
 628         struct lu_name          *lname;
 629         int                      rc;
 630         ENTRY;
 631
 632         DEBUG_REQ(D_INODE, req, "rename_tgt: insert (%s->"DFID") in "DFID,
 633                   rr->rr_tgt, PFID(rr->rr_fid2), PFID(rr->rr_fid1));
 634
 635         /* step 1: lookup & lock the tgt dir. */
 636         lh_tgtdir = &info->mti_lh[MDT_LH_PARENT];
 637         mdt_lock_pdo_init(lh_tgtdir, LCK_PW, rr->rr_tgt,
 638                           rr->rr_tgtlen);
 639         mtgtdir = mdt_object_find_lock(info, rr->rr_fid1, lh_tgtdir,
 640                                        MDS_INODELOCK_UPDATE);
 641         if (IS_ERR(mtgtdir))
 642                 RETURN(PTR_ERR(mtgtdir));
 643
 644         /* step 2: find & lock the target object if exists. */
 645         mdt_set_capainfo(info, 0, rr->rr_fid1, BYPASS_CAPA);
 646         lname = mdt_name(info->mti_env, (char *)rr->rr_tgt, rr->rr_tgtlen);
 647         rc = mdo_lookup(info->mti_env, mdt_object_child(mtgtdir),
 648                         lname, tgt_fid, &info->mti_spec);
 649         if (rc != 0 && rc != -ENOENT) {
 650                 GOTO(out_unlock_tgtdir, rc);
 651         } else if (rc == 0) {
 652                 /*
 653                  * In case of replay that name can be already inserted, check
 654                  * that and do nothing if so.
 655                  */
 656                 if (lu_fid_eq(tgt_fid, rr->rr_fid2))
 657                         GOTO(out_unlock_tgtdir, rc);
 658
 659                 lh_tgt = &info->mti_lh[MDT_LH_CHILD];
 660                 mdt_lock_reg_init(lh_tgt, LCK_EX);
 661
 662                 mtgt = mdt_object_find_lock(info, tgt_fid, lh_tgt,
 663                                             MDS_INODELOCK_LOOKUP);
 664                 if (IS_ERR(mtgt))
 665                         GOTO(out_unlock_tgtdir, rc = PTR_ERR(mtgt));
 666
 667                 mdt_reint_init_ma(info, ma);
 668                 if (!ma->ma_lmm || !ma->ma_cookie)
 669                         GOTO(out_unlock_tgt, rc = -EINVAL);
 670
 671                 rc = mdo_rename_tgt(info->mti_env, mdt_object_child(mtgtdir),
 672                                     mdt_object_child(mtgt), rr->rr_fid2,
 673                                     lname, ma);
 674         } else /* -ENOENT */ {
 675                 rc = mdo_name_insert(info->mti_env, mdt_object_child(mtgtdir),
 676                                      lname, rr->rr_fid2, ma);
 677         }
 678
 679         /* handle last link of tgt object */
 680         if (rc == 0 && mtgt)
 681                 mdt_handle_last_unlink(info, mtgt, ma);
 682
 683         EXIT;
 684 out_unlock_tgt:
 685         if (mtgt)
 686                 mdt_object_unlock_put(info, mtgt, lh_tgt, rc);
 687 out_unlock_tgtdir:
 688         mdt_object_unlock_put(info, mtgtdir, lh_tgtdir, rc);
 689         return rc;
 690 }
 691
 692 static int mdt_rename_lock(struct mdt_thread_info *info,
 693                            struct lustre_handle *lh)
 694 {
 695         struct ldlm_namespace *ns     = info->mti_mdt->mdt_namespace;
 696         ldlm_policy_data_t    *policy = &info->mti_policy;
 697         struct ldlm_res_id    *res_id = &info->mti_res_id;
 698         struct lu_site        *ls;
 699         int rc;
 700         ENTRY;
 701
 702         /*
 703          * Disable global rename BFL lock temporarily because
 704          * when a mds do rename recoverying, which might enqueue
 705          * BFL lock to the controller mds. and this req might be
 706          * replay req for controller mds. but we did not have
 707          * such handling in controller mds. XXX
 708          */
 709         RETURN(0);
 710         ls = info->mti_mdt->mdt_md_dev.md_lu_dev.ld_site;
 711         fid_build_reg_res_name(&LUSTRE_BFL_FID, res_id);
 712
 713         memset(policy, 0, sizeof *policy);
 714         policy->l_inodebits.bits = MDS_INODELOCK_UPDATE;
 715
 716         if (ls->ls_control_exp == NULL) {
 717                 int flags = LDLM_FL_LOCAL_ONLY | LDLM_FL_ATOMIC_CB;
 718
 719                 /*
 720                  * Current node is controller, that is mdt0, where we should
 721                  * take BFL lock.
 722                  */
 723                 rc = ldlm_cli_enqueue_local(ns, res_id, LDLM_IBITS, policy,
 724                                             LCK_EX, &flags, ldlm_blocking_ast,
 725                                             ldlm_completion_ast, NULL, NULL, 0,
 726                                             NULL, lh);
 727         } else {
 728                 struct ldlm_enqueue_info einfo = { LDLM_IBITS, LCK_EX,
 729                         ldlm_blocking_ast, ldlm_completion_ast, NULL, NULL };
 730                 int flags = 0;
 731
 732                 /*
 733                  * This is the case mdt0 is remote node, issue DLM lock like
 734                  * other clients.
 735                  */
 736                 rc = ldlm_cli_enqueue(ls->ls_control_exp, NULL, &einfo, res_id,
 737                                       policy, &flags, NULL, 0, NULL, lh, 0);
 738         }
 739
 740         RETURN(rc);
 741 }
 742
 743 static void mdt_rename_unlock(struct lustre_handle *lh)
 744 {
 745         ENTRY;
 746         /* Disable global rename BFL lock temporarily. see above XXX*/
 747         EXIT;
 748         return;
 749         LASSERT(lustre_handle_is_used(lh));
 750         ldlm_lock_decref(lh, LCK_EX);
 751         EXIT;
 752 }
 753
 754 /*
 755  * This is is_subdir() variant, it is CMD if cmm forwards it to correct
 756  * target. Source should not be ancestor of target dir. May be other rename
 757  * checks can be moved here later.
 758  */
 759 static int mdt_rename_sanity(struct mdt_thread_info *info, struct lu_fid *fid)
 760 {
 761         struct mdt_reint_record *rr = &info->mti_rr;
 762         struct lu_fid dst_fid = *rr->rr_fid2;
 763         struct mdt_object *dst;
 764         int rc = 0;
 765         ENTRY;
 766
 767         do {
 768                 LASSERT(fid_is_sane(&dst_fid));
 769                 dst = mdt_object_find(info->mti_env, info->mti_mdt, &dst_fid);
 770                 if (!IS_ERR(dst)) {
 771                         rc = mdo_is_subdir(info->mti_env,
 772                                            mdt_object_child(dst), fid,
 773                                            &dst_fid);
 774                         mdt_object_put(info->mti_env, dst);
 775                         if (rc != -EREMOTE && rc < 0) {
 776                                 CERROR("Failed mdo_is_subdir(), rc %d\n", rc);
 777                         } else {
 778                                 /* check the found fid */
 779                                 if (lu_fid_eq(&dst_fid, fid))
 780                                         rc = -EINVAL;
 781                         }
 782                 } else {
 783                         rc = PTR_ERR(dst);
 784                 }
 785         } while (rc == -EREMOTE);
 786
 787         RETURN(rc);
 788 }
 789
 790 static int mdt_reint_rename(struct mdt_thread_info *info,
 791                             struct mdt_lock_handle *lhc)
 792 {
 793         struct mdt_reint_record *rr = &info->mti_rr;
 794         struct md_attr          *ma = &info->mti_attr;
 795         struct ptlrpc_request   *req = mdt_info_req(info);
 796         struct mdt_object       *msrcdir;
 797         struct mdt_object       *mtgtdir;
 798         struct mdt_object       *mold;
 799         struct mdt_object       *mnew = NULL;
 800         struct mdt_lock_handle  *lh_srcdirp;
 801         struct mdt_lock_handle  *lh_tgtdirp;
 802         struct mdt_lock_handle  *lh_oldp;
 803         struct mdt_lock_handle  *lh_newp;
 804         struct lu_fid           *old_fid = &info->mti_tmp_fid1;
 805         struct lu_fid           *new_fid = &info->mti_tmp_fid2;
 806         struct lustre_handle     rename_lh = { 0 };
 807         struct lu_name           slname = { 0 };
 808         struct lu_name          *lname;
 809         int                      rc;
 810         ENTRY;
 811
 812         if (info->mti_dlm_req)
 813                 ldlm_request_cancel(mdt_info_req(info), info->mti_dlm_req, 0);
 814
 815         if (info->mti_cross_ref) {
 816                 rc = mdt_reint_rename_tgt(info);
 817                 RETURN(rc);
 818         }
 819
 820         DEBUG_REQ(D_INODE, req, "rename "DFID"/%s to "DFID"/%s",
 821                   PFID(rr->rr_fid1), rr->rr_name,
 822                   PFID(rr->rr_fid2), rr->rr_tgt);
 823
 824         rc = mdt_rename_lock(info, &rename_lh);
 825         if (rc) {
 826                 CERROR("Can't lock FS for rename, rc %d\n", rc);
 827                 RETURN(rc);
 828         }
 829
 830         lh_newp = &info->mti_lh[MDT_LH_NEW];
 831
 832         /* step 1: lock the source dir. */
 833         lh_srcdirp = &info->mti_lh[MDT_LH_PARENT];
 834         mdt_lock_pdo_init(lh_srcdirp, LCK_PW, rr->rr_name,
 835                           rr->rr_namelen);
 836         msrcdir = mdt_object_find_lock(info, rr->rr_fid1, lh_srcdirp,
 837                                        MDS_INODELOCK_UPDATE);
 838         if (IS_ERR(msrcdir))
 839                 GOTO(out_rename_lock, rc = PTR_ERR(msrcdir));
 840
 841         /* step 2: find & lock the target dir. */
 842         lh_tgtdirp = &info->mti_lh[MDT_LH_CHILD];
 843         mdt_lock_pdo_init(lh_tgtdirp, LCK_PW, rr->rr_tgt,
 844                           rr->rr_tgtlen);
 845         if (lu_fid_eq(rr->rr_fid1, rr->rr_fid2)) {
 846                 mdt_object_get(info->mti_env, msrcdir);
 847                 mtgtdir = msrcdir;
 848         } else {
 849                 mtgtdir = mdt_object_find(info->mti_env, info->mti_mdt,
 850                                           rr->rr_fid2);
 851                 if (IS_ERR(mtgtdir))
 852                         GOTO(out_unlock_source, rc = PTR_ERR(mtgtdir));
 853
 854                 rc = mdt_object_exists(mtgtdir);
 855                 if (rc == 0)
 856                         GOTO(out_unlock_target, rc = -ESTALE);
 857                 else if (rc > 0) {
 858                         /* we lock the target dir if it is local */
 859                         rc = mdt_object_lock(info, mtgtdir, lh_tgtdirp,
 860                                              MDS_INODELOCK_UPDATE,
 861                                              MDT_LOCAL_LOCK);
 862                         if (rc != 0)
 863                                 GOTO(out_unlock_target, rc);
 864                 }
 865         }
 866
 867         /* step 3: find & lock the old object. */
 868         lname = mdt_name(info->mti_env, (char *)rr->rr_name, rr->rr_namelen);
 869         mdt_name_copy(&slname, lname);
 870         rc = mdo_lookup(info->mti_env, mdt_object_child(msrcdir),
 871                         &slname, old_fid, &info->mti_spec);
 872         if (rc != 0)
 873                 GOTO(out_unlock_target, rc);
 874
 875         if (lu_fid_eq(old_fid, rr->rr_fid1) || lu_fid_eq(old_fid, rr->rr_fid2))
 876                 GOTO(out_unlock_target, rc = -EINVAL);
 877
 878         mold = mdt_object_find(info->mti_env, info->mti_mdt, old_fid);
 879         if (IS_ERR(mold))
 880                 GOTO(out_unlock_target, rc = PTR_ERR(mold));
 881
 882         lh_oldp = &info->mti_lh[MDT_LH_OLD];
 883         mdt_lock_reg_init(lh_oldp, LCK_EX);
 884         rc = mdt_object_lock(info, mold, lh_oldp, MDS_INODELOCK_LOOKUP,
 885                              MDT_CROSS_LOCK);
 886         if (rc != 0) {
 887                 mdt_object_put(info->mti_env, mold);
 888                 GOTO(out_unlock_target, rc);
 889         }
 890         mdt_set_capainfo(info, 2, old_fid, BYPASS_CAPA);
 891
 892         /* step 4: find & lock the new object. */
 893         /* new target object may not exist now */
 894         lname = mdt_name(info->mti_env, (char *)rr->rr_tgt, rr->rr_tgtlen);
 895         rc = mdo_lookup(info->mti_env, mdt_object_child(mtgtdir),
 896                         lname, new_fid, &info->mti_spec);
 897         if (rc == 0) {
 898                 /* the new_fid should have been filled at this moment */
 899                 if (lu_fid_eq(old_fid, new_fid))
 900                        GOTO(out_unlock_old, rc);
 901
 902                 if (lu_fid_eq(new_fid, rr->rr_fid1) ||
 903                     lu_fid_eq(new_fid, rr->rr_fid2))
 904                         GOTO(out_unlock_old, rc = -EINVAL);
 905
 906                 mdt_lock_reg_init(lh_newp, LCK_EX);
 907                 mnew = mdt_object_find(info->mti_env, info->mti_mdt, new_fid);
 908                 if (IS_ERR(mnew))
 909                         GOTO(out_unlock_old, rc = PTR_ERR(mnew));
 910
 911                 rc = mdt_object_lock(info, mnew, lh_newp,
 912                                      MDS_INODELOCK_FULL, MDT_CROSS_LOCK);
 913                 if (rc != 0) {
 914                         mdt_object_put(info->mti_env, mnew);
 915                         GOTO(out_unlock_old, rc);
 916                 }
 917                 mdt_set_capainfo(info, 3, new_fid, BYPASS_CAPA);
 918         } else if (rc != -EREMOTE && rc != -ENOENT)
 919                 GOTO(out_unlock_old, rc);
 920
 921         /* step 5: rename it */
 922         mdt_reint_init_ma(info, ma);
 923         if (!ma->ma_lmm || !ma->ma_cookie)
 924                 GOTO(out_unlock_new, rc = -EINVAL);
 925
 926         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
 927                        OBD_FAIL_MDS_REINT_RENAME_WRITE);
 928
 929
 930         /* Check if @dst is subdir of @src. */
 931         rc = mdt_rename_sanity(info, old_fid);
 932         if (rc)
 933                 GOTO(out_unlock_new, rc);
 934
 935         rc = mdo_rename(info->mti_env, mdt_object_child(msrcdir),
 936                         mdt_object_child(mtgtdir), old_fid, &slname,
 937                         (mnew ? mdt_object_child(mnew) : NULL),
 938                         lname, ma);
 939
 940         /* handle last link of tgt object */
 941         if (rc == 0 && mnew)
 942                 mdt_handle_last_unlink(info, mnew, ma);
 943
 944         EXIT;
 945 out_unlock_new:
 946         if (mnew)
 947                 mdt_object_unlock_put(info, mnew, lh_newp, rc);
 948 out_unlock_old:
 949         mdt_object_unlock_put(info, mold, lh_oldp, rc);
 950 out_unlock_target:
 951         mdt_object_unlock_put(info, mtgtdir, lh_tgtdirp, rc);
 952 out_unlock_source:
 953         mdt_object_unlock_put(info, msrcdir, lh_srcdirp, rc);
 954 out_rename_lock:
 955         mdt_rename_unlock(&rename_lh);
 956         return rc;
 957 }
 958
 959 typedef int (*mdt_reinter)(struct mdt_thread_info *info,
 960                            struct mdt_lock_handle *lhc);
 961
 962 static mdt_reinter reinters[REINT_MAX] = {
 963         [REINT_SETATTR]  = mdt_reint_setattr,
 964         [REINT_CREATE]   = mdt_reint_create,
 965         [REINT_LINK]     = mdt_reint_link,
 966         [REINT_UNLINK]   = mdt_reint_unlink,
 967         [REINT_RENAME]   = mdt_reint_rename,
 968         [REINT_OPEN]     = mdt_reint_open,
 969         [REINT_SETXATTR] = mdt_reint_setxattr
 970 };
 971
 972 int mdt_reint_rec(struct mdt_thread_info *info,
 973                   struct mdt_lock_handle *lhc)
 974 {
 975         int rc;
 976         ENTRY;
 977
 978         rc = reinters[info->mti_rr.rr_opcode](info, lhc);
 979
 980         RETURN(rc);
 981 }