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