lustre/cmm/cmm_object.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 [sun.com URL with a
  20  * copy of GPLv2].
  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  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/cmm/cmm_object.c
  37  *
  38  * Lustre Cluster Metadata Manager (cmm)
  39  *
  40  * Author: Mike Pershin <tappro@clusterfs.com>
  41  */
  42
  43 #ifndef EXPORT_SYMTAB
  44 # define EXPORT_SYMTAB
  45 #endif
  46
  47 #define DEBUG_SUBSYSTEM S_MDS
  48
  49 #include <lustre_fid.h>
  50 #include "cmm_internal.h"
  51 #include "mdc_internal.h"
  52
  53 int cmm_fld_lookup(struct cmm_device *cm, const struct lu_fid *fid,
  54                    mdsno_t *mds, const struct lu_env *env)
  55 {
  56         int rc = 0;
  57         ENTRY;
  58
  59         LASSERT(fid_is_sane(fid));
  60
  61         rc = fld_client_lookup(cm->cmm_fld, fid_seq(fid), mds, env);
  62         if (rc) {
  63                 CERROR("Can't find mds by seq "LPX64", rc %d\n",
  64                        fid_seq(fid), rc);
  65                 RETURN(rc);
  66         }
  67
  68         if (*mds > cm->cmm_tgt_count) {
  69                 CERROR("Got invalid mdsno: "LPU64" (max: %u)\n",
  70                        *mds, cm->cmm_tgt_count);
  71                 rc = -EINVAL;
  72         } else {
  73                 CDEBUG(D_INFO, "CMM: got MDS "LPU64" for sequence: "
  74                        LPU64"\n", *mds, fid_seq(fid));
  75         }
  76
  77         RETURN (rc);
  78 }
  79
  80 static struct md_object_operations cml_mo_ops;
  81 static struct md_dir_operations    cml_dir_ops;
  82 static struct lu_object_operations cml_obj_ops;
  83
  84 static struct md_object_operations cmr_mo_ops;
  85 static struct md_dir_operations    cmr_dir_ops;
  86 static struct lu_object_operations cmr_obj_ops;
  87
  88 struct lu_object *cmm_object_alloc(const struct lu_env *env,
  89                                    const struct lu_object_header *loh,
  90                                    struct lu_device *ld)
  91 {
  92         const struct lu_fid *fid = &loh->loh_fid;
  93         struct lu_object  *lo = NULL;
  94         struct cmm_device *cd;
  95         mdsno_t mds;
  96         int rc = 0;
  97
  98         ENTRY;
  99
 100         cd = lu2cmm_dev(ld);
 101         if (cd->cmm_flags & CMM_INITIALIZED) {
 102                 /* get object location */
 103                 rc = cmm_fld_lookup(lu2cmm_dev(ld), fid, &mds, env);
 104                 if (rc)
 105                         RETURN(NULL);
 106         } else
 107                 /*
 108                  * Device is not yet initialized, cmm_object is being created
 109                  * as part of early bootstrap procedure (it is /ROOT, or /fld,
 110                  * etc.). Such object *has* to be local.
 111                  */
 112                 mds = cd->cmm_local_num;
 113
 114         /* select the proper set of operations based on object location */
 115         if (mds == cd->cmm_local_num) {
 116                 struct cml_object *clo;
 117
 118                 OBD_ALLOC_PTR(clo);
 119                 if (clo != NULL) {
 120                         lo = &clo->cmm_obj.cmo_obj.mo_lu;
 121                         lu_object_init(lo, NULL, ld);
 122                         clo->cmm_obj.cmo_obj.mo_ops = &cml_mo_ops;
 123                         clo->cmm_obj.cmo_obj.mo_dir_ops = &cml_dir_ops;
 124                         lo->lo_ops = &cml_obj_ops;
 125                 }
 126         } else {
 127                 struct cmr_object *cro;
 128
 129                 OBD_ALLOC_PTR(cro);
 130                 if (cro != NULL) {
 131                         lo = &cro->cmm_obj.cmo_obj.mo_lu;
 132                         lu_object_init(lo, NULL, ld);
 133                         cro->cmm_obj.cmo_obj.mo_ops = &cmr_mo_ops;
 134                         cro->cmm_obj.cmo_obj.mo_dir_ops = &cmr_dir_ops;
 135                         lo->lo_ops = &cmr_obj_ops;
 136                         cro->cmo_num = mds;
 137                 }
 138         }
 139         RETURN(lo);
 140 }
 141
 142 /*
 143  * CMM has two types of objects - local and remote. They have different set
 144  * of operations so we are avoiding multiple checks in code.
 145  */
 146
 147 /* get local child device */
 148 static struct lu_device *cml_child_dev(struct cmm_device *d)
 149 {
 150         return &d->cmm_child->md_lu_dev;
 151 }
 152
 153 /* lu_object operations */
 154 static void cml_object_free(const struct lu_env *env,
 155                             struct lu_object *lo)
 156 {
 157         struct cml_object *clo = lu2cml_obj(lo);
 158         lu_object_fini(lo);
 159         OBD_FREE_PTR(clo);
 160 }
 161
 162 static int cml_object_init(const struct lu_env *env, struct lu_object *lo)
 163 {
 164         struct cmm_device *cd = lu2cmm_dev(lo->lo_dev);
 165         struct lu_device  *c_dev;
 166         struct lu_object  *c_obj;
 167         int rc;
 168
 169         ENTRY;
 170
 171 #ifdef HAVE_SPLIT_SUPPORT
 172         if (cd->cmm_tgt_count == 0)
 173                 lu2cml_obj(lo)->clo_split = CMM_SPLIT_DENIED;
 174         else
 175                 lu2cml_obj(lo)->clo_split = CMM_SPLIT_UNKNOWN;
 176 #endif
 177         c_dev = cml_child_dev(cd);
 178         if (c_dev == NULL) {
 179                 rc = -ENOENT;
 180         } else {
 181                 c_obj = c_dev->ld_ops->ldo_object_alloc(env,
 182                                                         lo->lo_header, c_dev);
 183                 if (c_obj != NULL) {
 184                         lu_object_add(lo, c_obj);
 185                         rc = 0;
 186                 } else {
 187                         rc = -ENOMEM;
 188                 }
 189         }
 190
 191         RETURN(rc);
 192 }
 193
 194 static int cml_object_print(const struct lu_env *env, void *cookie,
 195                             lu_printer_t p, const struct lu_object *lo)
 196 {
 197         return (*p)(env, cookie, LUSTRE_CMM_NAME"-local@%p", lo);
 198 }
 199
 200 static struct lu_object_operations cml_obj_ops = {
 201         .loo_object_init    = cml_object_init,
 202         .loo_object_free    = cml_object_free,
 203         .loo_object_print   = cml_object_print
 204 };
 205
 206 /* CMM local md_object operations */
 207 static int cml_object_create(const struct lu_env *env,
 208                              struct md_object *mo,
 209                              const struct md_op_spec *spec,
 210                              struct md_attr *attr)
 211 {
 212         int rc;
 213         ENTRY;
 214         rc = mo_object_create(env, md_object_next(mo), spec, attr);
 215         RETURN(rc);
 216 }
 217
 218 static int cml_permission(const struct lu_env *env,
 219                           struct md_object *p, struct md_object *c,
 220                           struct md_attr *attr, int mask)
 221 {
 222         int rc;
 223         ENTRY;
 224         rc = mo_permission(env, md_object_next(p), md_object_next(c),
 225                            attr, mask);
 226         RETURN(rc);
 227 }
 228
 229 static int cml_attr_get(const struct lu_env *env, struct md_object *mo,
 230                         struct md_attr *attr)
 231 {
 232         int rc;
 233         ENTRY;
 234         rc = mo_attr_get(env, md_object_next(mo), attr);
 235         RETURN(rc);
 236 }
 237
 238 static int cml_attr_set(const struct lu_env *env, struct md_object *mo,
 239                         const struct md_attr *attr)
 240 {
 241         int rc;
 242         ENTRY;
 243         rc = mo_attr_set(env, md_object_next(mo), attr);
 244         RETURN(rc);
 245 }
 246
 247 static int cml_xattr_get(const struct lu_env *env, struct md_object *mo,
 248                          struct lu_buf *buf, const char *name)
 249 {
 250         int rc;
 251         ENTRY;
 252         rc = mo_xattr_get(env, md_object_next(mo), buf, name);
 253         RETURN(rc);
 254 }
 255
 256 static int cml_readlink(const struct lu_env *env, struct md_object *mo,
 257                         struct lu_buf *buf)
 258 {
 259         int rc;
 260         ENTRY;
 261         rc = mo_readlink(env, md_object_next(mo), buf);
 262         RETURN(rc);
 263 }
 264
 265 static int cml_xattr_list(const struct lu_env *env, struct md_object *mo,
 266                           struct lu_buf *buf)
 267 {
 268         int rc;
 269         ENTRY;
 270         rc = mo_xattr_list(env, md_object_next(mo), buf);
 271         RETURN(rc);
 272 }
 273
 274 static int cml_xattr_set(const struct lu_env *env, struct md_object *mo,
 275                          const struct lu_buf *buf, const char *name,
 276                          int fl)
 277 {
 278         int rc;
 279         ENTRY;
 280         rc = mo_xattr_set(env, md_object_next(mo), buf, name, fl);
 281         RETURN(rc);
 282 }
 283
 284 static int cml_xattr_del(const struct lu_env *env, struct md_object *mo,
 285                          const char *name)
 286 {
 287         int rc;
 288         ENTRY;
 289         rc = mo_xattr_del(env, md_object_next(mo), name);
 290         RETURN(rc);
 291 }
 292
 293 static int cml_ref_add(const struct lu_env *env, struct md_object *mo,
 294                        const struct md_attr *ma)
 295 {
 296         int rc;
 297         ENTRY;
 298         rc = mo_ref_add(env, md_object_next(mo), ma);
 299         RETURN(rc);
 300 }
 301
 302 static int cml_ref_del(const struct lu_env *env, struct md_object *mo,
 303                        struct md_attr *ma)
 304 {
 305         int rc;
 306         ENTRY;
 307         rc = mo_ref_del(env, md_object_next(mo), ma);
 308         RETURN(rc);
 309 }
 310
 311 static int cml_open(const struct lu_env *env, struct md_object *mo,
 312                     int flags)
 313 {
 314         int rc;
 315         ENTRY;
 316         rc = mo_open(env, md_object_next(mo), flags);
 317         RETURN(rc);
 318 }
 319
 320 static int cml_close(const struct lu_env *env, struct md_object *mo,
 321                      struct md_attr *ma)
 322 {
 323         int rc;
 324         ENTRY;
 325         rc = mo_close(env, md_object_next(mo), ma);
 326         RETURN(rc);
 327 }
 328
 329 static int cml_readpage(const struct lu_env *env, struct md_object *mo,
 330                         const struct lu_rdpg *rdpg)
 331 {
 332         int rc;
 333         ENTRY;
 334         rc = mo_readpage(env, md_object_next(mo), rdpg);
 335         RETURN(rc);
 336 }
 337
 338 static int cml_capa_get(const struct lu_env *env, struct md_object *mo,
 339                         struct lustre_capa *capa, int renewal)
 340 {
 341         int rc;
 342         ENTRY;
 343         rc = mo_capa_get(env, md_object_next(mo), capa, renewal);
 344         RETURN(rc);
 345 }
 346
 347 static struct md_object_operations cml_mo_ops = {
 348         .moo_permission    = cml_permission,
 349         .moo_attr_get      = cml_attr_get,
 350         .moo_attr_set      = cml_attr_set,
 351         .moo_xattr_get     = cml_xattr_get,
 352         .moo_xattr_list    = cml_xattr_list,
 353         .moo_xattr_set     = cml_xattr_set,
 354         .moo_xattr_del     = cml_xattr_del,
 355         .moo_object_create = cml_object_create,
 356         .moo_ref_add       = cml_ref_add,
 357         .moo_ref_del       = cml_ref_del,
 358         .moo_open          = cml_open,
 359         .moo_close         = cml_close,
 360         .moo_readpage      = cml_readpage,
 361         .moo_readlink      = cml_readlink,
 362         .moo_capa_get      = cml_capa_get
 363 };
 364
 365 /* md_dir operations */
 366 static int cml_lookup(const struct lu_env *env, struct md_object *mo_p,
 367                       const struct lu_name *lname, struct lu_fid *lf,
 368                       struct md_op_spec *spec)
 369 {
 370         int rc;
 371         ENTRY;
 372
 373 #ifdef HAVE_SPLIT_SUPPORT
 374         if (spec != NULL && spec->sp_ck_split) {
 375                 rc = cmm_split_check(env, mo_p, lname->ln_name);
 376                 if (rc)
 377                         RETURN(rc);
 378         }
 379 #endif
 380         rc = mdo_lookup(env, md_object_next(mo_p), lname, lf, spec);
 381         RETURN(rc);
 382
 383 }
 384
 385 static mdl_mode_t cml_lock_mode(const struct lu_env *env,
 386                                 struct md_object *mo, mdl_mode_t lm)
 387 {
 388         int rc = MDL_MINMODE;
 389         ENTRY;
 390
 391 #ifdef HAVE_SPLIT_SUPPORT
 392         rc = cmm_split_access(env, mo, lm);
 393 #endif
 394
 395         RETURN(rc);
 396 }
 397
 398 static int cml_create(const struct lu_env *env, struct md_object *mo_p,
 399                       const struct lu_name *lname, struct md_object *mo_c,
 400                       struct md_op_spec *spec, struct md_attr *ma)
 401 {
 402         int rc;
 403         ENTRY;
 404
 405 #ifdef HAVE_SPLIT_SUPPORT
 406         /* Lock mode always should be sane. */
 407         LASSERT(spec->sp_cr_mode != MDL_MINMODE);
 408
 409         /*
 410          * Sigh... This is long story. MDT may have race with detecting if split
 411          * is possible in cmm. We know this race and let it live, because
 412          * getting it rid (with some sem or spinlock) will also mean that
 413          * PDIROPS for create will not work because we kill parallel work, what
 414          * is really bad for performance and makes no sense having PDIROPS. So,
 415          * we better allow the race to live, but split dir only if some of
 416          * concurrent threads takes EX lock, not matter which one. So that, say,
 417          * two concurrent threads may have different lock modes on directory (CW
 418          * and EX) and not first one which comes here and see that split is
 419          * possible should split the dir, but only that one which has EX
 420          * lock. And we do not care that in this case, split may happen a bit
 421          * later (when dir size will not be necessarily 64K, but may be a bit
 422          * larger). So that, we allow concurrent creates and protect split by EX
 423          * lock.
 424          */
 425         if (spec->sp_cr_mode == MDL_EX) {
 426                 /*
 427                  * Try to split @mo_p. If split is ok, -ERESTART is returned and
 428                  * current thread will not peoceed with create. Instead it sends
 429                  * -ERESTART to client to let it know that correct MDT should be
 430                  * choosen.
 431                  */
 432                 rc = cmm_split_dir(env, mo_p);
 433                 if (rc)
 434                         /*
 435                          * -ERESTART or some split error is returned, we can't
 436                          * proceed with create.
 437                          */
 438                         GOTO(out, rc);
 439         }
 440
 441         if (spec != NULL && spec->sp_ck_split) {
 442                 /*
 443                  * Check for possible split directory and let caller know that
 444                  * it should tell client that directory is split and operation
 445                  * should repeat to correct MDT.
 446                  */
 447                 rc = cmm_split_check(env, mo_p, lname->ln_name);
 448                 if (rc)
 449                         GOTO(out, rc);
 450         }
 451 #endif
 452
 453         rc = mdo_create(env, md_object_next(mo_p), lname, md_object_next(mo_c),
 454                         spec, ma);
 455
 456         EXIT;
 457 #ifdef HAVE_SPLIT_SUPPORT
 458 out:
 459 #endif
 460         return rc;
 461 }
 462
 463 static int cml_create_data(const struct lu_env *env, struct md_object *p,
 464                            struct md_object *o,
 465                            const struct md_op_spec *spec,
 466                            struct md_attr *ma)
 467 {
 468         int rc;
 469         ENTRY;
 470         rc = mdo_create_data(env, md_object_next(p), md_object_next(o),
 471                              spec, ma);
 472         RETURN(rc);
 473 }
 474
 475 static int cml_link(const struct lu_env *env, struct md_object *mo_p,
 476                     struct md_object *mo_s, const struct lu_name *lname,
 477                     struct md_attr *ma)
 478 {
 479         int rc;
 480         ENTRY;
 481         rc = mdo_link(env, md_object_next(mo_p), md_object_next(mo_s),
 482                       lname, ma);
 483         RETURN(rc);
 484 }
 485
 486 static int cml_unlink(const struct lu_env *env, struct md_object *mo_p,
 487                       struct md_object *mo_c, const struct lu_name *lname,
 488                       struct md_attr *ma)
 489 {
 490         int rc;
 491         ENTRY;
 492         rc = mdo_unlink(env, md_object_next(mo_p), md_object_next(mo_c),
 493                         lname, ma);
 494         RETURN(rc);
 495 }
 496
 497 /* rename is split to local/remote by location of new parent dir */
 498 struct md_object *md_object_find(const struct lu_env *env,
 499                                  struct md_device *md,
 500                                  const struct lu_fid *f)
 501 {
 502         struct lu_object *o;
 503         struct md_object *m;
 504         ENTRY;
 505
 506         o = lu_object_find(env, md2lu_dev(md)->ld_site, f);
 507         if (IS_ERR(o))
 508                 m = (struct md_object *)o;
 509         else {
 510                 o = lu_object_locate(o->lo_header, md2lu_dev(md)->ld_type);
 511                 m = o ? lu2md(o) : NULL;
 512         }
 513         RETURN(m);
 514 }
 515
 516 static int cmm_mode_get(const struct lu_env *env, struct md_device *md,
 517                         const struct lu_fid *lf, struct md_attr *ma,
 518                         int *remote)
 519 {
 520         struct md_object *mo_s = md_object_find(env, md, lf);
 521         struct cmm_thread_info *cmi;
 522         struct md_attr *tmp_ma;
 523         int rc;
 524         ENTRY;
 525
 526         if (IS_ERR(mo_s))
 527                 RETURN(PTR_ERR(mo_s));
 528
 529         if (remote && (lu_object_exists(&mo_s->mo_lu) < 0))
 530                 *remote = 1;
 531
 532         cmi = cmm_env_info(env);
 533         tmp_ma = &cmi->cmi_ma;
 534         tmp_ma->ma_need = MA_INODE;
 535         tmp_ma->ma_valid = 0;
 536         /* get type from src, can be remote req */
 537         rc = mo_attr_get(env, md_object_next(mo_s), tmp_ma);
 538         if (rc == 0) {
 539                 ma->ma_attr.la_mode = tmp_ma->ma_attr.la_mode;
 540                 ma->ma_attr.la_uid = tmp_ma->ma_attr.la_uid;
 541                 ma->ma_attr.la_gid = tmp_ma->ma_attr.la_gid;
 542                 ma->ma_attr.la_flags = tmp_ma->ma_attr.la_flags;
 543                 ma->ma_attr.la_valid |= LA_MODE | LA_UID | LA_GID | LA_FLAGS;
 544         }
 545         lu_object_put(env, &mo_s->mo_lu);
 546         RETURN(rc);
 547 }
 548
 549 static int cmm_rename_ctime(const struct lu_env *env, struct md_device *md,
 550                             const struct lu_fid *lf, struct md_attr *ma)
 551 {
 552         struct md_object *mo_s = md_object_find(env, md, lf);
 553         int rc;
 554         ENTRY;
 555
 556         if (IS_ERR(mo_s))
 557                 RETURN(PTR_ERR(mo_s));
 558
 559         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
 560         /* set ctime to obj, can be remote req */
 561         rc = mo_attr_set(env, md_object_next(mo_s), ma);
 562         lu_object_put(env, &mo_s->mo_lu);
 563         RETURN(rc);
 564 }
 565
 566 static inline void cml_rename_warn(const char *fname,
 567                                   struct md_object *mo_po,
 568                                   struct md_object *mo_pn,
 569                                   const struct lu_fid *lf,
 570                                   const char *s_name,
 571                                   struct md_object *mo_t,
 572                                   const char *t_name,
 573                                   int err)
 574 {
 575         if (mo_t)
 576                 CWARN("cml_rename failed for %s, should revoke: [mo_po "DFID"] "
 577                       "[mo_pn "DFID"] [lf "DFID"] [sname %s] [mo_t "DFID"] "
 578                       "[tname %s] [err %d]\n", fname,
 579                       PFID(lu_object_fid(&mo_po->mo_lu)),
 580                       PFID(lu_object_fid(&mo_pn->mo_lu)),
 581                       PFID(lf), s_name,
 582                       PFID(lu_object_fid(&mo_t->mo_lu)),
 583                       t_name, err);
 584         else
 585                 CWARN("cml_rename failed for %s, should revoke: [mo_po "DFID"] "
 586                       "[mo_pn "DFID"] [lf "DFID"] [sname %s] [mo_t NULL] "
 587                       "[tname %s] [err %d]\n", fname,
 588                       PFID(lu_object_fid(&mo_po->mo_lu)),
 589                       PFID(lu_object_fid(&mo_pn->mo_lu)),
 590                       PFID(lf), s_name,
 591                       t_name, err);
 592 }
 593
 594 static int cml_rename(const struct lu_env *env, struct md_object *mo_po,
 595                       struct md_object *mo_pn, const struct lu_fid *lf,
 596                       const struct lu_name *ls_name, struct md_object *mo_t,
 597                       const struct lu_name *lt_name, struct md_attr *ma)
 598 {
 599         struct cmm_thread_info *cmi;
 600         struct md_attr *tmp_ma = NULL;
 601         struct md_object *tmp_t = mo_t;
 602         int remote = 0, rc;
 603         ENTRY;
 604
 605         rc = cmm_mode_get(env, md_obj2dev(mo_po), lf, ma, &remote);
 606         if (rc)
 607                 RETURN(rc);
 608
 609         if (mo_t && lu_object_exists(&mo_t->mo_lu) < 0) {
 610                 /* XXX: mo_t is remote object and there is RPC to unlink it.
 611                  * before that, do local sanity check for rename first. */
 612                 if (!remote) {
 613                         struct md_object *mo_s = md_object_find(env,
 614                                                         md_obj2dev(mo_po), lf);
 615                         if (IS_ERR(mo_s))
 616                                 RETURN(PTR_ERR(mo_s));
 617
 618                         LASSERT(lu_object_exists(&mo_s->mo_lu) > 0);
 619                         rc = mo_permission(env, md_object_next(mo_po),
 620                                            md_object_next(mo_s),
 621                                            ma, MAY_RENAME_SRC);
 622                         lu_object_put(env, &mo_s->mo_lu);
 623                         if (rc)
 624                                 RETURN(rc);
 625                 } else {
 626                         rc = mo_permission(env, NULL, md_object_next(mo_po),
 627                                            ma, MAY_UNLINK | MAY_VTX_FULL);
 628                         if (rc)
 629                                 RETURN(rc);
 630                 }
 631
 632                 rc = mo_permission(env, NULL, md_object_next(mo_pn), ma,
 633                                    MAY_UNLINK | MAY_VTX_PART);
 634                 if (rc)
 635                         RETURN(rc);
 636
 637                 /*
 638                  * XXX: @ma will be changed after mo_ref_del, but we will use
 639                  * it for mdo_rename later, so save it before mo_ref_del.
 640                  */
 641                 cmi = cmm_env_info(env);
 642                 tmp_ma = &cmi->cmi_ma;
 643                 *tmp_ma = *ma;
 644                 rc = mo_ref_del(env, md_object_next(mo_t), ma);
 645                 if (rc)
 646                         RETURN(rc);
 647
 648                 tmp_ma->ma_attr_flags |= MDS_PERM_BYPASS;
 649                 mo_t = NULL;
 650         }
 651
 652         /* XXX: for src on remote MDS case, change its ctime before local
 653          * rename. Firstly, do local sanity check for rename if necessary. */
 654         if (remote) {
 655                 if (!tmp_ma) {
 656                         rc = mo_permission(env, NULL, md_object_next(mo_po),
 657                                            ma, MAY_UNLINK | MAY_VTX_FULL);
 658                         if (rc)
 659                                 RETURN(rc);
 660
 661                         if (mo_t) {
 662                                 LASSERT(lu_object_exists(&mo_t->mo_lu) > 0);
 663                                 rc = mo_permission(env, md_object_next(mo_pn),
 664                                                    md_object_next(mo_t),
 665                                                    ma, MAY_RENAME_TAR);
 666                                 if (rc)
 667                                         RETURN(rc);
 668                         } else {
 669                                 int mask;
 670
 671                                 if (mo_po != mo_pn)
 672                                         mask = (S_ISDIR(ma->ma_attr.la_mode) ?
 673                                                 MAY_LINK : MAY_CREATE);
 674                                 else
 675                                         mask = MAY_CREATE;
 676                                 rc = mo_permission(env, NULL,
 677                                                    md_object_next(mo_pn),
 678                                                    NULL, mask);
 679                                 if (rc)
 680                                         RETURN(rc);
 681                         }
 682
 683                         ma->ma_attr_flags |= MDS_PERM_BYPASS;
 684                 } else {
 685                         LASSERT(tmp_ma->ma_attr_flags & MDS_PERM_BYPASS);
 686                 }
 687
 688                 rc = cmm_rename_ctime(env, md_obj2dev(mo_po), lf,
 689                                       tmp_ma ? tmp_ma : ma);
 690                 if (rc) {
 691                         /* TODO: revoke mo_t if necessary. */
 692                         cml_rename_warn("cmm_rename_ctime", mo_po,
 693                                         mo_pn, lf, ls_name->ln_name,
 694                                         tmp_t, lt_name->ln_name, rc);
 695                         RETURN(rc);
 696                 }
 697         }
 698
 699         /* local rename, mo_t can be NULL */
 700         rc = mdo_rename(env, md_object_next(mo_po),
 701                         md_object_next(mo_pn), lf, ls_name,
 702                         md_object_next(mo_t), lt_name, tmp_ma ? tmp_ma : ma);
 703         if (rc)
 704                 /* TODO: revoke all cml_rename */
 705                 cml_rename_warn("mdo_rename", mo_po, mo_pn, lf,
 706                                 ls_name->ln_name, tmp_t, lt_name->ln_name, rc);
 707
 708         RETURN(rc);
 709 }
 710
 711 static int cml_rename_tgt(const struct lu_env *env, struct md_object *mo_p,
 712                           struct md_object *mo_t, const struct lu_fid *lf,
 713                           const struct lu_name *lname, struct md_attr *ma)
 714 {
 715         int rc;
 716         ENTRY;
 717
 718         rc = mdo_rename_tgt(env, md_object_next(mo_p),
 719                             md_object_next(mo_t), lf, lname, ma);
 720         RETURN(rc);
 721 }
 722 /* used only in case of rename_tgt() when target is not exist */
 723 static int cml_name_insert(const struct lu_env *env, struct md_object *p,
 724                            const struct lu_name *lname, const struct lu_fid *lf,
 725                            const struct md_attr *ma)
 726 {
 727         int rc;
 728         ENTRY;
 729
 730         rc = mdo_name_insert(env, md_object_next(p), lname, lf, ma);
 731
 732         RETURN(rc);
 733 }
 734
 735 static int cmm_is_subdir(const struct lu_env *env, struct md_object *mo,
 736                          const struct lu_fid *fid, struct lu_fid *sfid)
 737 {
 738         struct cmm_thread_info *cmi;
 739         int rc;
 740         ENTRY;
 741
 742         cmi = cmm_env_info(env);
 743         rc = cmm_mode_get(env, md_obj2dev(mo), fid, &cmi->cmi_ma, NULL);
 744         if (rc)
 745                 RETURN(rc);
 746
 747         if (!S_ISDIR(cmi->cmi_ma.ma_attr.la_mode))
 748                 RETURN(0);
 749
 750         rc = mdo_is_subdir(env, md_object_next(mo), fid, sfid);
 751         RETURN(rc);
 752 }
 753
 754 static struct md_dir_operations cml_dir_ops = {
 755         .mdo_is_subdir   = cmm_is_subdir,
 756         .mdo_lookup      = cml_lookup,
 757         .mdo_lock_mode   = cml_lock_mode,
 758         .mdo_create      = cml_create,
 759         .mdo_link        = cml_link,
 760         .mdo_unlink      = cml_unlink,
 761         .mdo_name_insert = cml_name_insert,
 762         .mdo_rename      = cml_rename,
 763         .mdo_rename_tgt  = cml_rename_tgt,
 764         .mdo_create_data = cml_create_data
 765 };
 766
 767 /* -------------------------------------------------------------------
 768  * remote CMM object operations. cmr_...
 769  */
 770 static inline struct cmr_object *lu2cmr_obj(struct lu_object *o)
 771 {
 772         return container_of0(o, struct cmr_object, cmm_obj.cmo_obj.mo_lu);
 773 }
 774 static inline struct cmr_object *md2cmr_obj(struct md_object *mo)
 775 {
 776         return container_of0(mo, struct cmr_object, cmm_obj.cmo_obj);
 777 }
 778 static inline struct cmr_object *cmm2cmr_obj(struct cmm_object *co)
 779 {
 780         return container_of0(co, struct cmr_object, cmm_obj);
 781 }
 782
 783 /* get proper child device from MDCs */
 784 static struct lu_device *cmr_child_dev(struct cmm_device *d, __u32 num)
 785 {
 786         struct lu_device *next = NULL;
 787         struct mdc_device *mdc;
 788
 789         spin_lock(&d->cmm_tgt_guard);
 790         list_for_each_entry(mdc, &d->cmm_targets, mc_linkage) {
 791                 if (mdc->mc_num == num) {
 792                         next = mdc2lu_dev(mdc);
 793                         break;
 794                 }
 795         }
 796         spin_unlock(&d->cmm_tgt_guard);
 797         return next;
 798 }
 799
 800 /* lu_object operations */
 801 static void cmr_object_free(const struct lu_env *env,
 802                             struct lu_object *lo)
 803 {
 804         struct cmr_object *cro = lu2cmr_obj(lo);
 805         lu_object_fini(lo);
 806         OBD_FREE_PTR(cro);
 807 }
 808
 809 static int cmr_object_init(const struct lu_env *env, struct lu_object *lo)
 810 {
 811         struct cmm_device *cd = lu2cmm_dev(lo->lo_dev);
 812         struct lu_device  *c_dev;
 813         struct lu_object  *c_obj;
 814         int rc;
 815
 816         ENTRY;
 817
 818         c_dev = cmr_child_dev(cd, lu2cmr_obj(lo)->cmo_num);
 819         if (c_dev == NULL) {
 820                 rc = -ENOENT;
 821         } else {
 822                 c_obj = c_dev->ld_ops->ldo_object_alloc(env,
 823                                                         lo->lo_header, c_dev);
 824                 if (c_obj != NULL) {
 825                         lu_object_add(lo, c_obj);
 826                         rc = 0;
 827                 } else {
 828                         rc = -ENOMEM;
 829                 }
 830         }
 831
 832         RETURN(rc);
 833 }
 834
 835 static int cmr_object_print(const struct lu_env *env, void *cookie,
 836                             lu_printer_t p, const struct lu_object *lo)
 837 {
 838         return (*p)(env, cookie, LUSTRE_CMM_NAME"-remote@%p", lo);
 839 }
 840
 841 static struct lu_object_operations cmr_obj_ops = {
 842         .loo_object_init    = cmr_object_init,
 843         .loo_object_free    = cmr_object_free,
 844         .loo_object_print   = cmr_object_print
 845 };
 846
 847 /* CMM remote md_object operations. All are invalid */
 848 static int cmr_object_create(const struct lu_env *env,
 849                              struct md_object *mo,
 850                              const struct md_op_spec *spec,
 851                              struct md_attr *ma)
 852 {
 853         return -EFAULT;
 854 }
 855
 856 static int cmr_permission(const struct lu_env *env,
 857                           struct md_object *p, struct md_object *c,
 858                           struct md_attr *attr, int mask)
 859 {
 860         return -EREMOTE;
 861 }
 862
 863 static int cmr_attr_get(const struct lu_env *env, struct md_object *mo,
 864                         struct md_attr *attr)
 865 {
 866         return -EREMOTE;
 867 }
 868
 869 static int cmr_attr_set(const struct lu_env *env, struct md_object *mo,
 870                         const struct md_attr *attr)
 871 {
 872         return -EFAULT;
 873 }
 874
 875 static int cmr_xattr_get(const struct lu_env *env, struct md_object *mo,
 876                          struct lu_buf *buf, const char *name)
 877 {
 878         return -EFAULT;
 879 }
 880
 881 static int cmr_readlink(const struct lu_env *env, struct md_object *mo,
 882                         struct lu_buf *buf)
 883 {
 884         return -EFAULT;
 885 }
 886
 887 static int cmr_xattr_list(const struct lu_env *env, struct md_object *mo,
 888                           struct lu_buf *buf)
 889 {
 890         return -EFAULT;
 891 }
 892
 893 static int cmr_xattr_set(const struct lu_env *env, struct md_object *mo,
 894                          const struct lu_buf *buf, const char *name,
 895                          int fl)
 896 {
 897         return -EFAULT;
 898 }
 899
 900 static int cmr_xattr_del(const struct lu_env *env, struct md_object *mo,
 901                          const char *name)
 902 {
 903         return -EFAULT;
 904 }
 905
 906 static int cmr_ref_add(const struct lu_env *env, struct md_object *mo,
 907                        const struct md_attr *ma)
 908 {
 909         return -EFAULT;
 910 }
 911
 912 static int cmr_ref_del(const struct lu_env *env, struct md_object *mo,
 913                        struct md_attr *ma)
 914 {
 915         return -EFAULT;
 916 }
 917
 918 static int cmr_open(const struct lu_env *env, struct md_object *mo,
 919                     int flags)
 920 {
 921         return -EREMOTE;
 922 }
 923
 924 static int cmr_close(const struct lu_env *env, struct md_object *mo,
 925                      struct md_attr *ma)
 926 {
 927         return -EFAULT;
 928 }
 929
 930 static int cmr_readpage(const struct lu_env *env, struct md_object *mo,
 931                         const struct lu_rdpg *rdpg)
 932 {
 933         return -EREMOTE;
 934 }
 935
 936 static int cmr_capa_get(const struct lu_env *env, struct md_object *mo,
 937                         struct lustre_capa *capa, int renewal)
 938 {
 939         return -EFAULT;
 940 }
 941
 942 static struct md_object_operations cmr_mo_ops = {
 943         .moo_permission    = cmr_permission,
 944         .moo_attr_get      = cmr_attr_get,
 945         .moo_attr_set      = cmr_attr_set,
 946         .moo_xattr_get     = cmr_xattr_get,
 947         .moo_xattr_set     = cmr_xattr_set,
 948         .moo_xattr_list    = cmr_xattr_list,
 949         .moo_xattr_del     = cmr_xattr_del,
 950         .moo_object_create = cmr_object_create,
 951         .moo_ref_add       = cmr_ref_add,
 952         .moo_ref_del       = cmr_ref_del,
 953         .moo_open          = cmr_open,
 954         .moo_close         = cmr_close,
 955         .moo_readpage      = cmr_readpage,
 956         .moo_readlink      = cmr_readlink,
 957         .moo_capa_get      = cmr_capa_get
 958 };
 959
 960 /* remote part of md_dir operations */
 961 static int cmr_lookup(const struct lu_env *env, struct md_object *mo_p,
 962                       const struct lu_name *lname, struct lu_fid *lf,
 963                       struct md_op_spec *spec)
 964 {
 965         /*
 966          * This can happens while rename() If new parent is remote dir, lookup
 967          * will happen here.
 968          */
 969
 970         return -EREMOTE;
 971 }
 972
 973 static mdl_mode_t cmr_lock_mode(const struct lu_env *env,
 974                                 struct md_object *mo, mdl_mode_t lm)
 975 {
 976         return MDL_MINMODE;
 977 }
 978
 979 /*
 980  * All methods below are cross-ref by nature. They consist of remote call and
 981  * local operation. Due to future rollback functionality there are several
 982  * limitations for such methods:
 983  * 1) remote call should be done at first to do epoch negotiation between all
 984  * MDS involved and to avoid the RPC inside transaction.
 985  * 2) only one RPC can be sent - also due to epoch negotiation.
 986  * For more details see rollback HLD/DLD.
 987  */
 988 static int cmr_create(const struct lu_env *env, struct md_object *mo_p,
 989                       const struct lu_name *lchild_name, struct md_object *mo_c,
 990                       struct md_op_spec *spec,
 991                       struct md_attr *ma)
 992 {
 993         struct cmm_thread_info *cmi;
 994         struct md_attr *tmp_ma;
 995         int rc;
 996         ENTRY;
 997
 998         /* Make sure that name isn't exist before doing remote call. */
 999         rc = mdo_lookup(env, md_object_next(mo_p), lchild_name,
1000                         &cmm_env_info(env)->cmi_fid, NULL);
1001         if (rc == 0)
1002                 RETURN(-EEXIST);
1003         else if (rc != -ENOENT)
1004                 RETURN(rc);
1005
1006         /* check the SGID attr */
1007         cmi = cmm_env_info(env);
1008         LASSERT(cmi);
1009         tmp_ma = &cmi->cmi_ma;
1010         tmp_ma->ma_valid = 0;
1011         tmp_ma->ma_need = MA_INODE;
1012
1013 #ifdef CONFIG_FS_POSIX_ACL
1014         if (!S_ISLNK(ma->ma_attr.la_mode)) {
1015                 tmp_ma->ma_acl = cmi->cmi_xattr_buf;
1016                 tmp_ma->ma_acl_size = sizeof(cmi->cmi_xattr_buf);
1017                 tmp_ma->ma_need |= MA_ACL_DEF;
1018         }
1019 #endif
1020         rc = mo_attr_get(env, md_object_next(mo_p), tmp_ma);
1021         if (rc)
1022                 RETURN(rc);
1023
1024         if (tmp_ma->ma_attr.la_mode & S_ISGID) {
1025                 ma->ma_attr.la_gid = tmp_ma->ma_attr.la_gid;
1026                 if (S_ISDIR(ma->ma_attr.la_mode)) {
1027                         ma->ma_attr.la_mode |= S_ISGID;
1028                         ma->ma_attr.la_valid |= LA_MODE;
1029                 }
1030         }
1031
1032 #ifdef CONFIG_FS_POSIX_ACL
1033         if (tmp_ma->ma_valid & MA_ACL_DEF) {
1034                 spec->u.sp_ea.fid = spec->u.sp_pfid;
1035                 spec->u.sp_ea.eadata = tmp_ma->ma_acl;
1036                 spec->u.sp_ea.eadatalen = tmp_ma->ma_acl_size;
1037                 spec->sp_cr_flags |= MDS_CREATE_RMT_ACL;
1038         }
1039 #endif
1040
1041         /* Local permission check for name_insert before remote ops. */
1042         rc = mo_permission(env, NULL, md_object_next(mo_p), NULL,
1043                            (S_ISDIR(ma->ma_attr.la_mode) ?
1044                            MAY_LINK : MAY_CREATE));
1045         if (rc)
1046                 RETURN(rc);
1047
1048         /* Remote object creation and local name insert. */
1049         /*
1050          * XXX: @ma will be changed after mo_object_create, but we will use
1051          * it for mdo_name_insert later, so save it before mo_object_create.
1052          */
1053         *tmp_ma = *ma;
1054         rc = mo_object_create(env, md_object_next(mo_c), spec, ma);
1055         if (rc == 0) {
1056                 tmp_ma->ma_attr_flags |= MDS_PERM_BYPASS;
1057                 rc = mdo_name_insert(env, md_object_next(mo_p), lchild_name,
1058                                      lu_object_fid(&mo_c->mo_lu), tmp_ma);
1059                 if (unlikely(rc)) {
1060                         /* TODO: remove object mo_c on remote MDS */
1061                         CWARN("cmr_create failed, should revoke: [mo_p "DFID"]"
1062                               " [name %s] [mo_c "DFID"] [err %d]\n",
1063                               PFID(lu_object_fid(&mo_p->mo_lu)),
1064                               lchild_name->ln_name,
1065                               PFID(lu_object_fid(&mo_c->mo_lu)), rc);
1066                 }
1067         }
1068
1069         RETURN(rc);
1070 }
1071
1072 static int cmr_link(const struct lu_env *env, struct md_object *mo_p,
1073                     struct md_object *mo_s, const struct lu_name *lname,
1074                     struct md_attr *ma)
1075 {
1076         int rc;
1077         ENTRY;
1078
1079         /* Make sure that name isn't exist before doing remote call. */
1080         rc = mdo_lookup(env, md_object_next(mo_p), lname,
1081                         &cmm_env_info(env)->cmi_fid, NULL);
1082         if (rc == 0) {
1083                 rc = -EEXIST;
1084         } else if (rc == -ENOENT) {
1085                 /* Local permission check for name_insert before remote ops. */
1086                 rc = mo_permission(env, NULL, md_object_next(mo_p), NULL,
1087                                    MAY_CREATE);
1088                 if (rc)
1089                         RETURN(rc);
1090
1091                 rc = mo_ref_add(env, md_object_next(mo_s), ma);
1092                 if (rc == 0) {
1093                         ma->ma_attr_flags |= MDS_PERM_BYPASS;
1094                         rc = mdo_name_insert(env, md_object_next(mo_p), lname,
1095                                              lu_object_fid(&mo_s->mo_lu), ma);
1096                         if (unlikely(rc)) {
1097                                 /* TODO: ref_del from mo_s on remote MDS */
1098                                 CWARN("cmr_link failed, should revoke: "
1099                                       "[mo_p "DFID"] [mo_s "DFID"] "
1100                                       "[name %s] [err %d]\n",
1101                                       PFID(lu_object_fid(&mo_p->mo_lu)),
1102                                       PFID(lu_object_fid(&mo_s->mo_lu)),
1103                                       lname->ln_name, rc);
1104                         }
1105                 }
1106         }
1107         RETURN(rc);
1108 }
1109
1110 static int cmr_unlink(const struct lu_env *env, struct md_object *mo_p,
1111                       struct md_object *mo_c, const struct lu_name *lname,
1112                       struct md_attr *ma)
1113 {
1114         struct cmm_thread_info *cmi;
1115         struct md_attr *tmp_ma;
1116         int rc;
1117         ENTRY;
1118
1119         /* Local permission check for name_remove before remote ops. */
1120         rc = mo_permission(env, NULL, md_object_next(mo_p), ma,
1121                            MAY_UNLINK | MAY_VTX_PART);
1122         if (rc)
1123                 RETURN(rc);
1124
1125         /*
1126          * XXX: @ma will be changed after mo_ref_del, but we will use
1127          * it for mdo_name_remove later, so save it before mo_ref_del.
1128          */
1129         cmi = cmm_env_info(env);
1130         tmp_ma = &cmi->cmi_ma;
1131         *tmp_ma = *ma;
1132         rc = mo_ref_del(env, md_object_next(mo_c), ma);
1133         if (rc == 0) {
1134                 tmp_ma->ma_attr_flags |= MDS_PERM_BYPASS;
1135                 rc = mdo_name_remove(env, md_object_next(mo_p), lname, tmp_ma);
1136                 if (unlikely(rc)) {
1137                         /* TODO: ref_add to mo_c on remote MDS */
1138                         CWARN("cmr_unlink failed, should revoke: [mo_p "DFID"]"
1139                               " [mo_c "DFID"] [name %s] [err %d]\n",
1140                               PFID(lu_object_fid(&mo_p->mo_lu)),
1141                               PFID(lu_object_fid(&mo_c->mo_lu)),
1142                               lname->ln_name, rc);
1143                 }
1144         }
1145
1146         RETURN(rc);
1147 }
1148
1149 static inline void cmr_rename_warn(const char *fname,
1150                                   struct md_object *mo_po,
1151                                   struct md_object *mo_pn,
1152                                   const struct lu_fid *lf,
1153                                   const char *s_name,
1154                                   const char *t_name,
1155                                   int err)
1156 {
1157         CWARN("cmr_rename failed for %s, should revoke: "
1158               "[mo_po "DFID"] [mo_pn "DFID"] [lf "DFID"] "
1159               "[sname %s] [tname %s] [err %d]\n", fname,
1160               PFID(lu_object_fid(&mo_po->mo_lu)),
1161               PFID(lu_object_fid(&mo_pn->mo_lu)),
1162               PFID(lf), s_name, t_name, err);
1163 }
1164
1165 static int cmr_rename(const struct lu_env *env,
1166                       struct md_object *mo_po, struct md_object *mo_pn,
1167                       const struct lu_fid *lf, const struct lu_name *ls_name,
1168                       struct md_object *mo_t, const struct lu_name *lt_name,
1169                       struct md_attr *ma)
1170 {
1171         struct cmm_thread_info *cmi;
1172         struct md_attr *tmp_ma;
1173         int rc;
1174         ENTRY;
1175
1176         LASSERT(mo_t == NULL);
1177
1178         /* get real type of src */
1179         rc = cmm_mode_get(env, md_obj2dev(mo_po), lf, ma, NULL);
1180         if (rc)
1181                 RETURN(rc);
1182
1183         /* Local permission check for name_remove before remote ops. */
1184         rc = mo_permission(env, NULL, md_object_next(mo_po), ma,
1185                            MAY_UNLINK | MAY_VTX_FULL);
1186         if (rc)
1187                 RETURN(rc);
1188
1189         /*
1190          * XXX: @ma maybe changed after mdo_rename_tgt, but we will use it
1191          * for mdo_name_remove later, so save it before mdo_rename_tgt.
1192          */
1193         cmi = cmm_env_info(env);
1194         tmp_ma = &cmi->cmi_ma;
1195         *tmp_ma = *ma;
1196         /* the mo_pn is remote directory, so we cannot even know if there is
1197          * mo_t or not. Therefore mo_t is NULL here but remote server should do
1198          * lookup and process this further */
1199         rc = mdo_rename_tgt(env, md_object_next(mo_pn),
1200                             NULL/* mo_t */, lf, lt_name, ma);
1201         if (rc)
1202                 RETURN(rc);
1203
1204         tmp_ma->ma_attr_flags |= MDS_PERM_BYPASS;
1205
1206         /* src object maybe on remote MDS, do remote ops first. */
1207         rc = cmm_rename_ctime(env, md_obj2dev(mo_po), lf, tmp_ma);
1208         if (unlikely(rc)) {
1209                 /* TODO: revoke mdo_rename_tgt */
1210                 cmr_rename_warn("cmm_rename_ctime", mo_po, mo_pn, lf,
1211                                 ls_name->ln_name, lt_name->ln_name, rc);
1212                 RETURN(rc);
1213         }
1214
1215         /* only old name is removed localy */
1216         rc = mdo_name_remove(env, md_object_next(mo_po), ls_name, tmp_ma);
1217         if (unlikely(rc))
1218                 /* TODO: revoke all cmr_rename */
1219                 cmr_rename_warn("mdo_name_remove", mo_po, mo_pn, lf,
1220                                 ls_name->ln_name, lt_name->ln_name, rc);
1221
1222         RETURN(rc);
1223 }
1224
1225 /* part of cross-ref rename(). Used to insert new name in new parent
1226  * and unlink target */
1227 static int cmr_rename_tgt(const struct lu_env *env,
1228                           struct md_object *mo_p, struct md_object *mo_t,
1229                           const struct lu_fid *lf, const struct lu_name *lname,
1230                           struct md_attr *ma)
1231 {
1232         struct cmm_thread_info *cmi;
1233         struct md_attr *tmp_ma;
1234         int rc;
1235         ENTRY;
1236
1237         /* target object is remote one */
1238         /* Local permission check for rename_tgt before remote ops. */
1239         rc = mo_permission(env, NULL, md_object_next(mo_p), ma,
1240                            MAY_UNLINK | MAY_VTX_PART);
1241         if (rc)
1242                 RETURN(rc);
1243
1244         /*
1245          * XXX: @ma maybe changed after mo_ref_del, but we will use
1246          * it for mdo_rename_tgt later, so save it before mo_ref_del.
1247          */
1248         cmi = cmm_env_info(env);
1249         tmp_ma = &cmi->cmi_ma;
1250         *tmp_ma = *ma;
1251         rc = mo_ref_del(env, md_object_next(mo_t), ma);
1252         /* continue locally with name handling only */
1253         if (rc == 0) {
1254                 tmp_ma->ma_attr_flags |= MDS_PERM_BYPASS;
1255                 rc = mdo_rename_tgt(env, md_object_next(mo_p),
1256                                     NULL, lf, lname, tmp_ma);
1257                 if (unlikely(rc)) {
1258                         /* TODO: ref_add to mo_t on remote MDS */
1259                         CWARN("cmr_rename_tgt failed, should revoke: "
1260                               "[mo_p "DFID"] [mo_t "DFID"] [lf "DFID"] "
1261                               "[name %s] [err %d]\n",
1262                               PFID(lu_object_fid(&mo_p->mo_lu)),
1263                               PFID(lu_object_fid(&mo_t->mo_lu)),
1264                               PFID(lf),
1265                               lname->ln_name, rc);
1266                 }
1267         }
1268         RETURN(rc);
1269 }
1270
1271 static struct md_dir_operations cmr_dir_ops = {
1272         .mdo_is_subdir   = cmm_is_subdir,
1273         .mdo_lookup      = cmr_lookup,
1274         .mdo_lock_mode   = cmr_lock_mode,
1275         .mdo_create      = cmr_create,
1276         .mdo_link        = cmr_link,
1277         .mdo_unlink      = cmr_unlink,
1278         .mdo_rename      = cmr_rename,
1279         .mdo_rename_tgt  = cmr_rename_tgt,
1280 };