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