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 (c) 2007, 2010, Oracle and/or its affiliates. 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  * \ingroup cmm
  54  * Lookup MDS number \a mds by FID \a fid.
  55  *
  56  * \param fid FID of object to find MDS
  57  * \param mds mds number to return.
  58  */
  59 int cmm_fld_lookup(struct cmm_device *cm, const struct lu_fid *fid,
  60                    mdsno_t *mds, const struct lu_env *env)
  61 {
  62         int rc = 0;
  63         ENTRY;
  64
  65         LASSERT(fid_is_sane(fid));
  66
  67         rc = fld_client_lookup(cm->cmm_fld, fid_seq(fid), mds,
  68                                LU_SEQ_RANGE_MDT, env);
  69         if (rc) {
  70                 CERROR("Can't find mds by seq "LPX64", rc %d\n",
  71                        fid_seq(fid), rc);
  72                 RETURN(rc);
  73         }
  74
  75         if (*mds > cm->cmm_tgt_count) {
  76                 CERROR("Got invalid mdsno: %x (max: %x)\n",
  77                        *mds, cm->cmm_tgt_count);
  78                 rc = -EINVAL;
  79         } else {
  80                 CDEBUG(D_INFO, "CMM: got MDS %x for sequence: "
  81                        LPX64"\n", *mds, fid_seq(fid));
  82         }
  83
  84         RETURN (rc);
  85 }
  86
  87 /**
  88  * \addtogroup cml
  89  * @{
  90  */
  91 static const struct md_object_operations cml_mo_ops;
  92 static const struct md_dir_operations    cml_dir_ops;
  93 static const struct lu_object_operations cml_obj_ops;
  94
  95 static const struct md_object_operations cmr_mo_ops;
  96 static const struct md_dir_operations    cmr_dir_ops;
  97 static const struct lu_object_operations cmr_obj_ops;
  98
  99 /**
 100  * \ingroup cmm
 101  * Allocate CMM object.
 102  */
 103 struct lu_object *cmm_object_alloc(const struct lu_env *env,
 104                                    const struct lu_object_header *loh,
 105                                    struct lu_device *ld)
 106 {
 107         const struct lu_fid *fid = &loh->loh_fid;
 108         struct lu_object  *lo = NULL;
 109         struct cmm_device *cd;
 110         mdsno_t mds;
 111         int rc = 0;
 112
 113         ENTRY;
 114
 115         cd = lu2cmm_dev(ld);
 116         if (cd->cmm_flags & CMM_INITIALIZED) {
 117                 /* get object location */
 118                 rc = cmm_fld_lookup(lu2cmm_dev(ld), fid, &mds, env);
 119                 if (rc)
 120                         RETURN(NULL);
 121         } else
 122                 /*
 123                  * Device is not yet initialized, cmm_object is being created
 124                  * as part of early bootstrap procedure (it is /ROOT, or /fld,
 125                  * etc.). Such object *has* to be local.
 126                  */
 127                 mds = cd->cmm_local_num;
 128
 129         /* select the proper set of operations based on object location */
 130         if (mds == cd->cmm_local_num) {
 131                 struct cml_object *clo;
 132
 133                 OBD_ALLOC_PTR(clo);
 134                 if (clo != NULL) {
 135                         lo = &clo->cmm_obj.cmo_obj.mo_lu;
 136                         lu_object_init(lo, NULL, ld);
 137                         clo->cmm_obj.cmo_obj.mo_ops = &cml_mo_ops;
 138                         clo->cmm_obj.cmo_obj.mo_dir_ops = &cml_dir_ops;
 139                         lo->lo_ops = &cml_obj_ops;
 140                 }
 141         } else {
 142                 struct cmr_object *cro;
 143
 144                 OBD_ALLOC_PTR(cro);
 145                 if (cro != NULL) {
 146                         lo = &cro->cmm_obj.cmo_obj.mo_lu;
 147                         lu_object_init(lo, NULL, ld);
 148                         cro->cmm_obj.cmo_obj.mo_ops = &cmr_mo_ops;
 149                         cro->cmm_obj.cmo_obj.mo_dir_ops = &cmr_dir_ops;
 150                         lo->lo_ops = &cmr_obj_ops;
 151                         cro->cmo_num = mds;
 152                 }
 153         }
 154         RETURN(lo);
 155 }
 156
 157 /**
 158  * Get local child device.
 159  */
 160 static struct lu_device *cml_child_dev(struct cmm_device *d)
 161 {
 162         return &d->cmm_child->md_lu_dev;
 163 }
 164
 165 /**
 166  * Free cml_object.
 167  */
 168 static void cml_object_free(const struct lu_env *env,
 169                             struct lu_object *lo)
 170 {
 171         struct cml_object *clo = lu2cml_obj(lo);
 172         lu_object_fini(lo);
 173         OBD_FREE_PTR(clo);
 174 }
 175
 176 /**
 177  * Initialize cml_object.
 178  */
 179 static int cml_object_init(const struct lu_env *env, struct lu_object *lo,
 180                            const struct lu_object_conf *unused)
 181 {
 182         struct cmm_device *cd = lu2cmm_dev(lo->lo_dev);
 183         struct lu_device  *c_dev;
 184         struct lu_object  *c_obj;
 185         int rc;
 186
 187         ENTRY;
 188
 189 #ifdef HAVE_SPLIT_SUPPORT
 190         if (cd->cmm_tgt_count == 0)
 191                 lu2cml_obj(lo)->clo_split = CMM_SPLIT_DENIED;
 192         else
 193                 lu2cml_obj(lo)->clo_split = CMM_SPLIT_UNKNOWN;
 194 #endif
 195         c_dev = cml_child_dev(cd);
 196         if (c_dev == NULL) {
 197                 rc = -ENOENT;
 198         } else {
 199                 c_obj = c_dev->ld_ops->ldo_object_alloc(env,
 200                                                         lo->lo_header, c_dev);
 201                 if (c_obj != NULL) {
 202                         lu_object_add(lo, c_obj);
 203                         rc = 0;
 204                 } else {
 205                         rc = -ENOMEM;
 206                 }
 207         }
 208
 209         RETURN(rc);
 210 }
 211
 212 static int cml_object_print(const struct lu_env *env, void *cookie,
 213                             lu_printer_t p, const struct lu_object *lo)
 214 {
 215         return (*p)(env, cookie, "[local]");
 216 }
 217
 218 static const struct lu_object_operations cml_obj_ops = {
 219         .loo_object_init    = cml_object_init,
 220         .loo_object_free    = cml_object_free,
 221         .loo_object_print   = cml_object_print
 222 };
 223
 224 /**
 225  * \name CMM local md_object operations.
 226  * All of them call just corresponding operations on next layer.
 227  * @{
 228  */
 229 static int cml_object_create(const struct lu_env *env,
 230                              struct md_object *mo,
 231                              const struct md_op_spec *spec,
 232                              struct md_attr *attr)
 233 {
 234         int rc;
 235         ENTRY;
 236         rc = mo_object_create(env, md_object_next(mo), spec, attr);
 237         RETURN(rc);
 238 }
 239
 240 static int cml_permission(const struct lu_env *env,
 241                           struct md_object *p, struct md_object *c,
 242                           struct md_attr *attr, int mask)
 243 {
 244         int rc;
 245         ENTRY;
 246         rc = mo_permission(env, md_object_next(p), md_object_next(c),
 247                            attr, mask);
 248         RETURN(rc);
 249 }
 250
 251 static int cml_attr_get(const struct lu_env *env, struct md_object *mo,
 252                         struct md_attr *attr)
 253 {
 254         int rc;
 255         ENTRY;
 256         rc = mo_attr_get(env, md_object_next(mo), attr);
 257         RETURN(rc);
 258 }
 259
 260 static int cml_attr_set(const struct lu_env *env, struct md_object *mo,
 261                         const struct md_attr *attr)
 262 {
 263         int rc;
 264         ENTRY;
 265         rc = mo_attr_set(env, md_object_next(mo), attr);
 266         RETURN(rc);
 267 }
 268
 269 static int cml_xattr_get(const struct lu_env *env, struct md_object *mo,
 270                          struct lu_buf *buf, const char *name)
 271 {
 272         int rc;
 273         ENTRY;
 274         rc = mo_xattr_get(env, md_object_next(mo), buf, name);
 275         RETURN(rc);
 276 }
 277
 278 static int cml_readlink(const struct lu_env *env, struct md_object *mo,
 279                         struct lu_buf *buf)
 280 {
 281         int rc;
 282         ENTRY;
 283         rc = mo_readlink(env, md_object_next(mo), buf);
 284         RETURN(rc);
 285 }
 286
 287 static int cml_changelog(const struct lu_env *env, enum changelog_rec_type type,
 288                          int flags, struct md_object *mo)
 289 {
 290         int rc;
 291         ENTRY;
 292         rc = mo_changelog(env, type, flags, md_object_next(mo));
 293         RETURN(rc);
 294 }
 295
 296 static int cml_xattr_list(const struct lu_env *env, struct md_object *mo,
 297                           struct lu_buf *buf)
 298 {
 299         int rc;
 300         ENTRY;
 301         rc = mo_xattr_list(env, md_object_next(mo), buf);
 302         RETURN(rc);
 303 }
 304
 305 static int cml_xattr_set(const struct lu_env *env, struct md_object *mo,
 306                          const struct lu_buf *buf, const char *name,
 307                          int fl)
 308 {
 309         int rc;
 310         ENTRY;
 311         rc = mo_xattr_set(env, md_object_next(mo), buf, name, fl);
 312         RETURN(rc);
 313 }
 314
 315 static int cml_xattr_del(const struct lu_env *env, struct md_object *mo,
 316                          const char *name)
 317 {
 318         int rc;
 319         ENTRY;
 320         rc = mo_xattr_del(env, md_object_next(mo), name);
 321         RETURN(rc);
 322 }
 323
 324 static int cml_ref_add(const struct lu_env *env, struct md_object *mo,
 325                        const struct md_attr *ma)
 326 {
 327         int rc;
 328         ENTRY;
 329         rc = mo_ref_add(env, md_object_next(mo), ma);
 330         RETURN(rc);
 331 }
 332
 333 static int cml_ref_del(const struct lu_env *env, struct md_object *mo,
 334                        struct md_attr *ma)
 335 {
 336         int rc;
 337         ENTRY;
 338         rc = mo_ref_del(env, md_object_next(mo), ma);
 339         RETURN(rc);
 340 }
 341
 342 static int cml_open(const struct lu_env *env, struct md_object *mo,
 343                     int flags)
 344 {
 345         int rc;
 346         ENTRY;
 347         rc = mo_open(env, md_object_next(mo), flags);
 348         RETURN(rc);
 349 }
 350
 351 static int cml_close(const struct lu_env *env, struct md_object *mo,
 352                      struct md_attr *ma, int mode)
 353 {
 354         int rc;
 355         ENTRY;
 356         rc = mo_close(env, md_object_next(mo), ma, mode);
 357         RETURN(rc);
 358 }
 359
 360 static int cml_readpage(const struct lu_env *env, struct md_object *mo,
 361                         const struct lu_rdpg *rdpg)
 362 {
 363         int rc;
 364         ENTRY;
 365         rc = mo_readpage(env, md_object_next(mo), rdpg);
 366         RETURN(rc);
 367 }
 368
 369 static int cml_capa_get(const struct lu_env *env, struct md_object *mo,
 370                         struct lustre_capa *capa, int renewal)
 371 {
 372         int rc;
 373         ENTRY;
 374         rc = mo_capa_get(env, md_object_next(mo), capa, renewal);
 375         RETURN(rc);
 376 }
 377
 378 static int cml_path(const struct lu_env *env, struct md_object *mo,
 379                     char *path, int pathlen, __u64 *recno, int *linkno)
 380 {
 381         int rc;
 382         ENTRY;
 383         rc = mo_path(env, md_object_next(mo), path, pathlen, recno, linkno);
 384         RETURN(rc);
 385 }
 386
 387 static int cml_file_lock(const struct lu_env *env, struct md_object *mo,
 388                          struct lov_mds_md *lmm, struct ldlm_extent *extent,
 389                          struct lustre_handle *lockh)
 390 {
 391         int rc;
 392         ENTRY;
 393         rc = mo_file_lock(env, md_object_next(mo), lmm, extent, lockh);
 394         RETURN(rc);
 395 }
 396
 397 static int cml_file_unlock(const struct lu_env *env, struct md_object *mo,
 398                            struct lov_mds_md *lmm, struct lustre_handle *lockh)
 399 {
 400         int rc;
 401         ENTRY;
 402         rc = mo_file_unlock(env, md_object_next(mo), lmm, lockh);
 403         RETURN(rc);
 404 }
 405
 406 static int cml_object_sync(const struct lu_env *env, struct md_object *mo)
 407 {
 408         int rc;
 409         ENTRY;
 410         rc = mo_object_sync(env, md_object_next(mo));
 411         RETURN(rc);
 412 }
 413
 414 static const struct md_object_operations cml_mo_ops = {
 415         .moo_permission    = cml_permission,
 416         .moo_attr_get      = cml_attr_get,
 417         .moo_attr_set      = cml_attr_set,
 418         .moo_xattr_get     = cml_xattr_get,
 419         .moo_xattr_list    = cml_xattr_list,
 420         .moo_xattr_set     = cml_xattr_set,
 421         .moo_xattr_del     = cml_xattr_del,
 422         .moo_object_create = cml_object_create,
 423         .moo_ref_add       = cml_ref_add,
 424         .moo_ref_del       = cml_ref_del,
 425         .moo_open          = cml_open,
 426         .moo_close         = cml_close,
 427         .moo_readpage      = cml_readpage,
 428         .moo_readlink      = cml_readlink,
 429         .moo_changelog     = cml_changelog,
 430         .moo_capa_get      = cml_capa_get,
 431         .moo_object_sync   = cml_object_sync,
 432         .moo_path          = cml_path,
 433         .moo_file_lock     = cml_file_lock,
 434         .moo_file_unlock   = cml_file_unlock,
 435 };
 436 /** @} */
 437
 438 /**
 439  * \name CMM local md_dir_operations.
 440  * @{
 441  */
 442 /**
 443  * cml lookup object fid by name.
 444  * This returns only FID by name.
 445  */
 446 static int cml_lookup(const struct lu_env *env, struct md_object *mo_p,
 447                       const struct lu_name *lname, struct lu_fid *lf,
 448                       struct md_op_spec *spec)
 449 {
 450         int rc;
 451         ENTRY;
 452
 453 #ifdef HAVE_SPLIT_SUPPORT
 454         if (spec != NULL && spec->sp_ck_split) {
 455                 rc = cmm_split_check(env, mo_p, lname->ln_name);
 456                 if (rc)
 457                         RETURN(rc);
 458         }
 459 #endif
 460         rc = mdo_lookup(env, md_object_next(mo_p), lname, lf, spec);
 461         RETURN(rc);
 462
 463 }
 464
 465 /**
 466  * Helper to return lock mode. Used in split cases only.
 467  */
 468 static mdl_mode_t cml_lock_mode(const struct lu_env *env,
 469                                 struct md_object *mo, mdl_mode_t lm)
 470 {
 471         int rc = MDL_MINMODE;
 472         ENTRY;
 473
 474 #ifdef HAVE_SPLIT_SUPPORT
 475         rc = cmm_split_access(env, mo, lm);
 476 #endif
 477
 478         RETURN(rc);
 479 }
 480
 481 /**
 482  * Create operation for cml.
 483  * Objects are local, but split can happen.
 484  * If split is not needed this will call next layer mdo_create().
 485  *
 486  * \param mo_p Parent directory. Local object.
 487  * \param lname name of file to create.
 488  * \param mo_c Child object. It has no real inode yet.
 489  * \param spec creation specification.
 490  * \param ma child object attributes.
 491  */
 492 static int cml_create(const struct lu_env *env, struct md_object *mo_p,
 493                       const struct lu_name *lname, struct md_object *mo_c,
 494                       struct md_op_spec *spec, struct md_attr *ma)
 495 {
 496         int rc;
 497         ENTRY;
 498
 499 #ifdef HAVE_SPLIT_SUPPORT
 500         /* Lock mode always should be sane. */
 501         LASSERT(spec->sp_cr_mode != MDL_MINMODE);
 502
 503         /*
 504          * Sigh... This is long story. MDT may have race with detecting if split
 505          * is possible in cmm. We know this race and let it live, because
 506          * getting it rid (with some sem or spinlock) will also mean that
 507          * PDIROPS for create will not work because we kill parallel work, what
 508          * is really bad for performance and makes no sense having PDIROPS. So,
 509          * we better allow the race to live, but split dir only if some of
 510          * concurrent threads takes EX lock, not matter which one. So that, say,
 511          * two concurrent threads may have different lock modes on directory (CW
 512          * and EX) and not first one which comes here and see that split is
 513          * possible should split the dir, but only that one which has EX
 514          * lock. And we do not care that in this case, split may happen a bit
 515          * later (when dir size will not be necessarily 64K, but may be a bit
 516          * larger). So that, we allow concurrent creates and protect split by EX
 517          * lock.
 518          */
 519         if (spec->sp_cr_mode == MDL_EX) {
 520                 /**
 521                  * Split cases:
 522                  * - Try to split \a mo_p upon each create operation.
 523                  *   If split is ok, -ERESTART is returned and current thread
 524                  *   will not peoceed with create. Instead it sends -ERESTART
 525                  *   to client to let it know that correct MDT must be chosen.
 526                  * \see cmm_split_dir()
 527                  */
 528                 rc = cmm_split_dir(env, mo_p);
 529                 if (rc)
 530                         /*
 531                          * -ERESTART or some split error is returned, we can't
 532                          * proceed with create.
 533                          */
 534                         GOTO(out, rc);
 535         }
 536
 537         if (spec != NULL && spec->sp_ck_split) {
 538                 /**
 539                  * - Directory is split already. Let the caller know that
 540                  * it should tell client that directory is split and operation
 541                  * should repeat to correct MDT.
 542                  * \see cmm_split_check()
 543                  */
 544                 rc = cmm_split_check(env, mo_p, lname->ln_name);
 545                 if (rc)
 546                         GOTO(out, rc);
 547         }
 548 #endif
 549
 550         rc = mdo_create(env, md_object_next(mo_p), lname, md_object_next(mo_c),
 551                         spec, ma);
 552
 553         EXIT;
 554 #ifdef HAVE_SPLIT_SUPPORT
 555 out:
 556 #endif
 557         return rc;
 558 }
 559
 560 /** Call mdo_create_data() on next layer. All objects are local. */
 561 static int cml_create_data(const struct lu_env *env, struct md_object *p,
 562                            struct md_object *o,
 563                            const struct md_op_spec *spec,
 564                            struct md_attr *ma)
 565 {
 566         int rc;
 567         ENTRY;
 568         rc = mdo_create_data(env, md_object_next(p), md_object_next(o),
 569                              spec, ma);
 570         RETURN(rc);
 571 }
 572
 573 /** Call mdo_link() on next layer. All objects are local. */
 574 static int cml_link(const struct lu_env *env, struct md_object *mo_p,
 575                     struct md_object *mo_s, const struct lu_name *lname,
 576                     struct md_attr *ma)
 577 {
 578         int rc;
 579         ENTRY;
 580         rc = mdo_link(env, md_object_next(mo_p), md_object_next(mo_s),
 581                       lname, ma);
 582         RETURN(rc);
 583 }
 584
 585 /** Call mdo_unlink() on next layer. All objects are local. */
 586 static int cml_unlink(const struct lu_env *env, struct md_object *mo_p,
 587                       struct md_object *mo_c, const struct lu_name *lname,
 588                       struct md_attr *ma)
 589 {
 590         int rc;
 591         ENTRY;
 592         rc = mdo_unlink(env, md_object_next(mo_p), md_object_next(mo_c),
 593                         lname, ma);
 594         RETURN(rc);
 595 }
 596
 597 /** Call mdo_lum_lmm_cmp() on next layer */
 598 static int cml_lum_lmm_cmp(const struct lu_env *env, struct md_object *mo_c,
 599                            const struct md_op_spec *spec, struct md_attr *ma)
 600 {
 601         int rc;
 602         ENTRY;
 603
 604         rc = mdo_lum_lmm_cmp(env, md_object_next(mo_c), spec, ma);
 605         RETURN(rc);
 606 }
 607
 608 /**
 609  * \ingroup cmm
 610  * Get mode of object.
 611  * Used in both cml and cmr hence can produce RPC to another server.
 612  */
 613 static int cmm_mode_get(const struct lu_env *env, struct md_device *md,
 614                         const struct lu_fid *lf, struct md_attr *ma,
 615                         int *remote)
 616 {
 617         struct md_object *mo_s = md_object_find_slice(env, md, lf);
 618         struct cmm_thread_info *cmi;
 619         struct md_attr *tmp_ma;
 620         int rc;
 621         ENTRY;
 622
 623         if (IS_ERR(mo_s))
 624                 RETURN(PTR_ERR(mo_s));
 625
 626         if (remote && (lu_object_exists(&mo_s->mo_lu) < 0))
 627                 *remote = 1;
 628
 629         cmi = cmm_env_info(env);
 630         tmp_ma = &cmi->cmi_ma;
 631         tmp_ma->ma_need = MA_INODE;
 632         tmp_ma->ma_valid = 0;
 633         /* get type from src, can be remote req */
 634         rc = mo_attr_get(env, md_object_next(mo_s), tmp_ma);
 635         if (rc == 0) {
 636                 ma->ma_attr.la_mode = tmp_ma->ma_attr.la_mode;
 637                 ma->ma_attr.la_uid = tmp_ma->ma_attr.la_uid;
 638                 ma->ma_attr.la_gid = tmp_ma->ma_attr.la_gid;
 639                 ma->ma_attr.la_flags = tmp_ma->ma_attr.la_flags;
 640                 ma->ma_attr.la_valid |= LA_MODE | LA_UID | LA_GID | LA_FLAGS;
 641         }
 642         lu_object_put(env, &mo_s->mo_lu);
 643         RETURN(rc);
 644 }
 645
 646 /**
 647  * \ingroup cmm
 648  * Set ctime for object.
 649  * Used in both cml and cmr hence can produce RPC to another server.
 650  */
 651 static int cmm_rename_ctime(const struct lu_env *env, struct md_device *md,
 652                             const struct lu_fid *lf, struct md_attr *ma)
 653 {
 654         struct md_object *mo_s = md_object_find_slice(env, md, lf);
 655         int rc;
 656         ENTRY;
 657
 658         if (IS_ERR(mo_s))
 659                 RETURN(PTR_ERR(mo_s));
 660
 661         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
 662         /* set ctime to obj, can be remote req */
 663         rc = mo_attr_set(env, md_object_next(mo_s), ma);
 664         lu_object_put(env, &mo_s->mo_lu);
 665         RETURN(rc);
 666 }
 667
 668 /** Helper to output debug information about rename operation. */
 669 static inline void cml_rename_warn(const char *fname,
 670                                   struct md_object *mo_po,
 671                                   struct md_object *mo_pn,
 672                                   const struct lu_fid *lf,
 673                                   const char *s_name,
 674                                   struct md_object *mo_t,
 675                                   const char *t_name,
 676                                   int err)
 677 {
 678         if (mo_t)
 679                 CWARN("cml_rename failed for %s, should revoke: [mo_po "DFID"] "
 680                       "[mo_pn "DFID"] [lf "DFID"] [sname %s] [mo_t "DFID"] "
 681                       "[tname %s] [err %d]\n", fname,
 682                       PFID(lu_object_fid(&mo_po->mo_lu)),
 683                       PFID(lu_object_fid(&mo_pn->mo_lu)),
 684                       PFID(lf), s_name,
 685                       PFID(lu_object_fid(&mo_t->mo_lu)),
 686                       t_name, err);
 687         else
 688                 CWARN("cml_rename failed for %s, should revoke: [mo_po "DFID"] "
 689                       "[mo_pn "DFID"] [lf "DFID"] [sname %s] [mo_t NULL] "
 690                       "[tname %s] [err %d]\n", fname,
 691                       PFID(lu_object_fid(&mo_po->mo_lu)),
 692                       PFID(lu_object_fid(&mo_pn->mo_lu)),
 693                       PFID(lf), s_name,
 694                       t_name, err);
 695 }
 696
 697 /**
 698  * Rename operation for cml.
 699  *
 700  * This is the most complex cross-reference operation. It may consist of up to 4
 701  * MDS server and require several RPCs to be sent.
 702  *
 703  * \param mo_po Old parent object.
 704  * \param mo_pn New parent object.
 705  * \param lf FID of object to rename.
 706  * \param ls_name Source file name.
 707  * \param mo_t target object. Should be NULL here.
 708  * \param lt_name Name of target file.
 709  * \param ma object attributes.
 710  */
 711 static int cml_rename(const struct lu_env *env, struct md_object *mo_po,
 712                       struct md_object *mo_pn, const struct lu_fid *lf,
 713                       const struct lu_name *ls_name, struct md_object *mo_t,
 714                       const struct lu_name *lt_name, struct md_attr *ma)
 715 {
 716         struct cmm_thread_info *cmi;
 717         struct md_attr *tmp_ma = NULL;
 718         struct md_object *tmp_t = mo_t;
 719         int remote = 0, rc;
 720         ENTRY;
 721
 722         rc = cmm_mode_get(env, md_obj2dev(mo_po), lf, ma, &remote);
 723         if (rc)
 724                 RETURN(rc);
 725
 726         if (mo_t && lu_object_exists(&mo_t->mo_lu) < 0) {
 727                 /**
 728                  * \note \a mo_t is remote object and there is RPC to unlink it.
 729                  * Before that, do local sanity check for rename first.
 730                  */
 731                 if (!remote) {
 732                         struct md_object *mo_s = md_object_find_slice(env,
 733                                                         md_obj2dev(mo_po), lf);
 734                         if (IS_ERR(mo_s))
 735                                 RETURN(PTR_ERR(mo_s));
 736
 737                         LASSERT(lu_object_exists(&mo_s->mo_lu) > 0);
 738                         rc = mo_permission(env, md_object_next(mo_po),
 739                                            md_object_next(mo_s),
 740                                            ma, MAY_RENAME_SRC);
 741                         lu_object_put(env, &mo_s->mo_lu);
 742                         if (rc)
 743                                 RETURN(rc);
 744                 } else {
 745                         rc = mo_permission(env, NULL, md_object_next(mo_po),
 746                                            ma, MAY_UNLINK | MAY_VTX_FULL);
 747                         if (rc)
 748                                 RETURN(rc);
 749                 }
 750
 751                 rc = mo_permission(env, NULL, md_object_next(mo_pn), ma,
 752                                    MAY_UNLINK | MAY_VTX_PART);
 753                 if (rc)
 754                         RETURN(rc);
 755
 756                 /*
 757                  * /note \a ma will be changed after mo_ref_del(), but we will use
 758                  * it for mdo_rename() later, so save it before mo_ref_del().
 759                  */
 760                 cmi = cmm_env_info(env);
 761                 tmp_ma = &cmi->cmi_ma;
 762                 *tmp_ma = *ma;
 763                 rc = mo_ref_del(env, md_object_next(mo_t), ma);
 764                 if (rc)
 765                         RETURN(rc);
 766
 767                 tmp_ma->ma_attr_flags |= MDS_PERM_BYPASS;
 768                 mo_t = NULL;
 769         }
 770
 771         /**
 772          * \note for src on remote MDS case, change its ctime before local
 773          * rename. Firstly, do local sanity check for rename if necessary.
 774          */
 775         if (remote) {
 776                 if (!tmp_ma) {
 777                         rc = mo_permission(env, NULL, md_object_next(mo_po),
 778                                            ma, MAY_UNLINK | MAY_VTX_FULL);
 779                         if (rc)
 780                                 RETURN(rc);
 781
 782                         if (mo_t) {
 783                                 LASSERT(lu_object_exists(&mo_t->mo_lu) > 0);
 784                                 rc = mo_permission(env, md_object_next(mo_pn),
 785                                                    md_object_next(mo_t),
 786                                                    ma, MAY_RENAME_TAR);
 787                                 if (rc)
 788                                         RETURN(rc);
 789                         } else {
 790                                 int mask;
 791
 792                                 if (mo_po != mo_pn)
 793                                         mask = (S_ISDIR(ma->ma_attr.la_mode) ?
 794                                                 MAY_LINK : MAY_CREATE);
 795                                 else
 796                                         mask = MAY_CREATE;
 797                                 rc = mo_permission(env, NULL,
 798                                                    md_object_next(mo_pn),
 799                                                    NULL, mask);
 800                                 if (rc)
 801                                         RETURN(rc);
 802                         }
 803
 804                         ma->ma_attr_flags |= MDS_PERM_BYPASS;
 805                 } else {
 806                         LASSERT(tmp_ma->ma_attr_flags & MDS_PERM_BYPASS);
 807                 }
 808
 809                 rc = cmm_rename_ctime(env, md_obj2dev(mo_po), lf,
 810                                       tmp_ma ? tmp_ma : ma);
 811                 if (rc) {
 812                         /* TODO: revoke mo_t if necessary. */
 813                         cml_rename_warn("cmm_rename_ctime", mo_po,
 814                                         mo_pn, lf, ls_name->ln_name,
 815                                         tmp_t, lt_name->ln_name, rc);
 816                         RETURN(rc);
 817                 }
 818         }
 819
 820         /* local rename, mo_t can be NULL */
 821         rc = mdo_rename(env, md_object_next(mo_po),
 822                         md_object_next(mo_pn), lf, ls_name,
 823                         md_object_next(mo_t), lt_name, tmp_ma ? tmp_ma : ma);
 824         if (rc)
 825                 /* TODO: revoke all cml_rename */
 826                 cml_rename_warn("mdo_rename", mo_po, mo_pn, lf,
 827                                 ls_name->ln_name, tmp_t, lt_name->ln_name, rc);
 828
 829         RETURN(rc);
 830 }
 831
 832 /**
 833  * Rename target partial operation.
 834  * Used for cross-ref rename.
 835  */
 836 static int cml_rename_tgt(const struct lu_env *env, struct md_object *mo_p,
 837                           struct md_object *mo_t, const struct lu_fid *lf,
 838                           const struct lu_name *lname, struct md_attr *ma)
 839 {
 840         int rc;
 841         ENTRY;
 842
 843         rc = mdo_rename_tgt(env, md_object_next(mo_p),
 844                             md_object_next(mo_t), lf, lname, ma);
 845         RETURN(rc);
 846 }
 847
 848 /**
 849  * Name insert only operation.
 850  * used only in case of rename_tgt() when target doesn't exist.
 851  */
 852 static int cml_name_insert(const struct lu_env *env, struct md_object *p,
 853                            const struct lu_name *lname, const struct lu_fid *lf,
 854                            const struct md_attr *ma)
 855 {
 856         int rc;
 857         ENTRY;
 858
 859         rc = mdo_name_insert(env, md_object_next(p), lname, lf, ma);
 860
 861         RETURN(rc);
 862 }
 863
 864 /**
 865  * \ingroup cmm
 866  * Check two fids are not subdirectories.
 867  */
 868 static int cmm_is_subdir(const struct lu_env *env, struct md_object *mo,
 869                          const struct lu_fid *fid, struct lu_fid *sfid)
 870 {
 871         struct cmm_thread_info *cmi;
 872         int rc;
 873         ENTRY;
 874
 875         cmi = cmm_env_info(env);
 876         rc = cmm_mode_get(env, md_obj2dev(mo), fid, &cmi->cmi_ma, NULL);
 877         if (rc)
 878                 RETURN(rc);
 879
 880         if (!S_ISDIR(cmi->cmi_ma.ma_attr.la_mode))
 881                 RETURN(0);
 882
 883         rc = mdo_is_subdir(env, md_object_next(mo), fid, sfid);
 884         RETURN(rc);
 885 }
 886
 887 static const struct md_dir_operations cml_dir_ops = {
 888         .mdo_is_subdir   = cmm_is_subdir,
 889         .mdo_lookup      = cml_lookup,
 890         .mdo_lock_mode   = cml_lock_mode,
 891         .mdo_create      = cml_create,
 892         .mdo_link        = cml_link,
 893         .mdo_unlink      = cml_unlink,
 894         .mdo_lum_lmm_cmp = cml_lum_lmm_cmp,
 895         .mdo_name_insert = cml_name_insert,
 896         .mdo_rename      = cml_rename,
 897         .mdo_rename_tgt  = cml_rename_tgt,
 898         .mdo_create_data = cml_create_data,
 899 };
 900 /** @} */
 901 /** @} */
 902
 903 /**
 904  * \addtogroup cmr
 905  * @{
 906  */
 907 /**
 908  * \name cmr helpers
 909  * @{
 910  */
 911 /** Get cmr_object from lu_object. */
 912 static inline struct cmr_object *lu2cmr_obj(struct lu_object *o)
 913 {
 914         return container_of0(o, struct cmr_object, cmm_obj.cmo_obj.mo_lu);
 915 }
 916 /** Get cmr_object from md_object. */
 917 static inline struct cmr_object *md2cmr_obj(struct md_object *mo)
 918 {
 919         return container_of0(mo, struct cmr_object, cmm_obj.cmo_obj);
 920 }
 921 /** Get cmr_object from cmm_object. */
 922 static inline struct cmr_object *cmm2cmr_obj(struct cmm_object *co)
 923 {
 924         return container_of0(co, struct cmr_object, cmm_obj);
 925 }
 926 /** @} */
 927
 928 /**
 929  * Get proper child device from MDCs.
 930  */
 931 static struct lu_device *cmr_child_dev(struct cmm_device *d, __u32 num)
 932 {
 933         struct lu_device *next = NULL;
 934         struct mdc_device *mdc;
 935
 936         cfs_spin_lock(&d->cmm_tgt_guard);
 937         cfs_list_for_each_entry(mdc, &d->cmm_targets, mc_linkage) {
 938                 if (mdc->mc_num == num) {
 939                         next = mdc2lu_dev(mdc);
 940                         break;
 941                 }
 942         }
 943         cfs_spin_unlock(&d->cmm_tgt_guard);
 944         return next;
 945 }
 946
 947 /**
 948  * Free cmr_object.
 949  */
 950 static void cmr_object_free(const struct lu_env *env,
 951                             struct lu_object *lo)
 952 {
 953         struct cmr_object *cro = lu2cmr_obj(lo);
 954         lu_object_fini(lo);
 955         OBD_FREE_PTR(cro);
 956 }
 957
 958 /**
 959  * Initialize cmr object.
 960  */
 961 static int cmr_object_init(const struct lu_env *env, struct lu_object *lo,
 962                            const struct lu_object_conf *unused)
 963 {
 964         struct cmm_device *cd = lu2cmm_dev(lo->lo_dev);
 965         struct lu_device  *c_dev;
 966         struct lu_object  *c_obj;
 967         int rc;
 968
 969         ENTRY;
 970
 971         c_dev = cmr_child_dev(cd, lu2cmr_obj(lo)->cmo_num);
 972         if (c_dev == NULL) {
 973                 rc = -ENOENT;
 974         } else {
 975                 c_obj = c_dev->ld_ops->ldo_object_alloc(env,
 976                                                         lo->lo_header, c_dev);
 977                 if (c_obj != NULL) {
 978                         lu_object_add(lo, c_obj);
 979                         rc = 0;
 980                 } else {
 981                         rc = -ENOMEM;
 982                 }
 983         }
 984
 985         RETURN(rc);
 986 }
 987
 988 /**
 989  * Output lu_object data.
 990  */
 991 static int cmr_object_print(const struct lu_env *env, void *cookie,
 992                             lu_printer_t p, const struct lu_object *lo)
 993 {
 994         const struct cmr_object *cro = lu2cmr_obj((struct lu_object *)lo);
 995         return (*p)(env, cookie, "[remote](mds_num=%d)", cro->cmo_num);
 996 }
 997
 998 /**
 999  * Cmr instance of lu_object_operations.
1000  */
1001 static const struct lu_object_operations cmr_obj_ops = {
1002         .loo_object_init    = cmr_object_init,
1003         .loo_object_free    = cmr_object_free,
1004         .loo_object_print   = cmr_object_print
1005 };
1006
1007 /**
1008  * \name cmr remote md_object operations.
1009  * All operations here are invalid and return errors. There is no local object
1010  * so these operations return two kinds of error:
1011  * -# -EFAULT if operation is prohibited.
1012  * -# -EREMOTE if operation can be done just to notify upper level about remote
1013  *  object.
1014  *
1015  * @{
1016  */
1017 static int cmr_object_create(const struct lu_env *env,
1018                              struct md_object *mo,
1019                              const struct md_op_spec *spec,
1020                              struct md_attr *ma)
1021 {
1022         return -EFAULT;
1023 }
1024
1025 static int cmr_permission(const struct lu_env *env,
1026                           struct md_object *p, struct md_object *c,
1027                           struct md_attr *attr, int mask)
1028 {
1029         return -EREMOTE;
1030 }
1031
1032 static int cmr_attr_get(const struct lu_env *env, struct md_object *mo,
1033                         struct md_attr *attr)
1034 {
1035         return -EREMOTE;
1036 }
1037
1038 static int cmr_attr_set(const struct lu_env *env, struct md_object *mo,
1039                         const struct md_attr *attr)
1040 {
1041         return -EFAULT;
1042 }
1043
1044 static int cmr_xattr_get(const struct lu_env *env, struct md_object *mo,
1045                          struct lu_buf *buf, const char *name)
1046 {
1047         return -EFAULT;
1048 }
1049
1050 static int cmr_readlink(const struct lu_env *env, struct md_object *mo,
1051                         struct lu_buf *buf)
1052 {
1053         return -EFAULT;
1054 }
1055
1056 static int cmr_changelog(const struct lu_env *env, enum changelog_rec_type type,
1057                          int flags, struct md_object *mo)
1058 {
1059         return -EFAULT;
1060 }
1061
1062 static int cmr_xattr_list(const struct lu_env *env, struct md_object *mo,
1063                           struct lu_buf *buf)
1064 {
1065         return -EFAULT;
1066 }
1067
1068 static int cmr_xattr_set(const struct lu_env *env, struct md_object *mo,
1069                          const struct lu_buf *buf, const char *name,
1070                          int fl)
1071 {
1072         return -EFAULT;
1073 }
1074
1075 static int cmr_xattr_del(const struct lu_env *env, struct md_object *mo,
1076                          const char *name)
1077 {
1078         return -EFAULT;
1079 }
1080
1081 static int cmr_ref_add(const struct lu_env *env, struct md_object *mo,
1082                        const struct md_attr *ma)
1083 {
1084         return -EFAULT;
1085 }
1086
1087 static int cmr_ref_del(const struct lu_env *env, struct md_object *mo,
1088                        struct md_attr *ma)
1089 {
1090         return -EFAULT;
1091 }
1092
1093 static int cmr_open(const struct lu_env *env, struct md_object *mo,
1094                     int flags)
1095 {
1096         return -EREMOTE;
1097 }
1098
1099 static int cmr_close(const struct lu_env *env, struct md_object *mo,
1100                      struct md_attr *ma, int mode)
1101 {
1102         return -EFAULT;
1103 }
1104
1105 static int cmr_readpage(const struct lu_env *env, struct md_object *mo,
1106                         const struct lu_rdpg *rdpg)
1107 {
1108         return -EREMOTE;
1109 }
1110
1111 static int cmr_capa_get(const struct lu_env *env, struct md_object *mo,
1112                         struct lustre_capa *capa, int renewal)
1113 {
1114         return -EFAULT;
1115 }
1116
1117 static int cmr_path(const struct lu_env *env, struct md_object *obj,
1118                     char *path, int pathlen, __u64 *recno, int *linkno)
1119 {
1120         return -EREMOTE;
1121 }
1122
1123 static int cmr_object_sync(const struct lu_env *env, struct md_object *mo)
1124 {
1125         return -EFAULT;
1126 }
1127
1128 static int cmr_file_lock(const struct lu_env *env, struct md_object *mo,
1129                          struct lov_mds_md *lmm, struct ldlm_extent *extent,
1130                          struct lustre_handle *lockh)
1131 {
1132         return -EREMOTE;
1133 }
1134
1135 static int cmr_file_unlock(const struct lu_env *env, struct md_object *mo,
1136                            struct lov_mds_md *lmm, struct lustre_handle *lockh)
1137 {
1138         return -EREMOTE;
1139 }
1140
1141 static int cmr_lum_lmm_cmp(const struct lu_env *env, struct md_object *mo_c,
1142                            const struct md_op_spec *spec, struct md_attr *ma)
1143 {
1144         return -EREMOTE;
1145 }
1146
1147 /** Set of md_object_operations for cmr. */
1148 static const struct md_object_operations cmr_mo_ops = {
1149         .moo_permission    = cmr_permission,
1150         .moo_attr_get      = cmr_attr_get,
1151         .moo_attr_set      = cmr_attr_set,
1152         .moo_xattr_get     = cmr_xattr_get,
1153         .moo_xattr_set     = cmr_xattr_set,
1154         .moo_xattr_list    = cmr_xattr_list,
1155         .moo_xattr_del     = cmr_xattr_del,
1156         .moo_object_create = cmr_object_create,
1157         .moo_ref_add       = cmr_ref_add,
1158         .moo_ref_del       = cmr_ref_del,
1159         .moo_open          = cmr_open,
1160         .moo_close         = cmr_close,
1161         .moo_readpage      = cmr_readpage,
1162         .moo_readlink      = cmr_readlink,
1163         .moo_changelog     = cmr_changelog,
1164         .moo_capa_get      = cmr_capa_get,
1165         .moo_object_sync   = cmr_object_sync,
1166         .moo_path          = cmr_path,
1167         .moo_file_lock     = cmr_file_lock,
1168         .moo_file_unlock   = cmr_file_unlock,
1169 };
1170 /** @} */
1171
1172 /**
1173  * \name cmr md_dir operations.
1174  *
1175  * All methods below are cross-ref by nature. They consist of remote call and
1176  * local operation. Due to future rollback functionality there are several
1177  * limitations for such methods:
1178  * -# remote call should be done at first to do epoch negotiation between all
1179  * MDS involved and to avoid the RPC inside transaction.
1180  * -# only one RPC can be sent - also due to epoch negotiation.
1181  * For more details see rollback HLD/DLD.
1182  * @{
1183  */
1184 static int cmr_lookup(const struct lu_env *env, struct md_object *mo_p,
1185                       const struct lu_name *lname, struct lu_fid *lf,
1186                       struct md_op_spec *spec)
1187 {
1188         /*
1189          * This can happens while rename() If new parent is remote dir, lookup
1190          * will happen here.
1191          */
1192
1193         return -EREMOTE;
1194 }
1195
1196 /** Return lock mode. */
1197 static mdl_mode_t cmr_lock_mode(const struct lu_env *env,
1198                                 struct md_object *mo, mdl_mode_t lm)
1199 {
1200         return MDL_MINMODE;
1201 }
1202
1203 /**
1204  * Create operation for cmr.
1205  * Remote object creation and local name insert.
1206  *
1207  * \param mo_p Parent directory. Local object.
1208  * \param lchild_name name of file to create.
1209  * \param mo_c Child object. It has no real inode yet.
1210  * \param spec creation specification.
1211  * \param ma child object attributes.
1212  */
1213 static int cmr_create(const struct lu_env *env, struct md_object *mo_p,
1214                       const struct lu_name *lchild_name, struct md_object *mo_c,
1215                       struct md_op_spec *spec,
1216                       struct md_attr *ma)
1217 {
1218         struct cmm_thread_info *cmi;
1219         struct md_attr *tmp_ma;
1220         int rc;
1221         ENTRY;
1222
1223         /* Make sure that name isn't exist before doing remote call. */
1224         rc = mdo_lookup(env, md_object_next(mo_p), lchild_name,
1225                         &cmm_env_info(env)->cmi_fid, NULL);
1226         if (rc == 0)
1227                 RETURN(-EEXIST);
1228         else if (rc != -ENOENT)
1229                 RETURN(rc);
1230
1231         /* check the SGID attr */
1232         cmi = cmm_env_info(env);
1233         LASSERT(cmi);
1234         tmp_ma = &cmi->cmi_ma;
1235         tmp_ma->ma_valid = 0;
1236         tmp_ma->ma_need = MA_INODE;
1237
1238 #ifdef CONFIG_FS_POSIX_ACL
1239         if (!S_ISLNK(ma->ma_attr.la_mode)) {
1240                 tmp_ma->ma_acl = cmi->cmi_xattr_buf;
1241                 tmp_ma->ma_acl_size = sizeof(cmi->cmi_xattr_buf);
1242                 tmp_ma->ma_need |= MA_ACL_DEF;
1243         }
1244 #endif
1245         rc = mo_attr_get(env, md_object_next(mo_p), tmp_ma);
1246         if (rc)
1247                 RETURN(rc);
1248
1249         if (tmp_ma->ma_attr.la_mode & S_ISGID) {
1250                 ma->ma_attr.la_gid = tmp_ma->ma_attr.la_gid;
1251                 if (S_ISDIR(ma->ma_attr.la_mode)) {
1252                         ma->ma_attr.la_mode |= S_ISGID;
1253                         ma->ma_attr.la_valid |= LA_MODE;
1254                 }
1255         }
1256
1257 #ifdef CONFIG_FS_POSIX_ACL
1258         if (tmp_ma->ma_valid & MA_ACL_DEF) {
1259                 spec->u.sp_ea.fid = spec->u.sp_pfid;
1260                 spec->u.sp_ea.eadata = tmp_ma->ma_acl;
1261                 spec->u.sp_ea.eadatalen = tmp_ma->ma_acl_size;
1262                 spec->sp_cr_flags |= MDS_CREATE_RMT_ACL;
1263         }
1264 #endif
1265
1266         /* Local permission check for name_insert before remote ops. */
1267         rc = mo_permission(env, NULL, md_object_next(mo_p), NULL,
1268                            (S_ISDIR(ma->ma_attr.la_mode) ?
1269                            MAY_LINK : MAY_CREATE));
1270         if (rc)
1271                 RETURN(rc);
1272
1273         /**
1274          * \note \a ma will be changed after mo_object_create(), but we will use
1275          * it for mdo_name_insert() later, so save it before mo_object_create().
1276          */
1277         *tmp_ma = *ma;
1278         rc = mo_object_create(env, md_object_next(mo_c), spec, ma);
1279         if (rc == 0) {
1280                 tmp_ma->ma_attr_flags |= MDS_PERM_BYPASS;
1281                 rc = mdo_name_insert(env, md_object_next(mo_p), lchild_name,
1282                                      lu_object_fid(&mo_c->mo_lu), tmp_ma);
1283                 if (unlikely(rc)) {
1284                         /* TODO: remove object mo_c on remote MDS */
1285                         CWARN("cmr_create failed, should revoke: [mo_p "DFID"]"
1286                               " [name %s] [mo_c "DFID"] [err %d]\n",
1287                               PFID(lu_object_fid(&mo_p->mo_lu)),
1288                               lchild_name->ln_name,
1289                               PFID(lu_object_fid(&mo_c->mo_lu)), rc);
1290                 }
1291         }
1292
1293         RETURN(rc);
1294 }
1295
1296 /**
1297  * Link operations for cmr.
1298  *
1299  * The link RPC is always issued to the server where source parent is living.
1300  * The first operation to do is object nlink increment on remote server.
1301  * Second one is local mdo_name_insert().
1302  *
1303  * \param mo_p parent directory. It is local.
1304  * \param mo_s source object to link. It is remote.
1305  * \param lname Name of link file.
1306  * \param ma object attributes.
1307  */
1308 static int cmr_link(const struct lu_env *env, struct md_object *mo_p,
1309                     struct md_object *mo_s, const struct lu_name *lname,
1310                     struct md_attr *ma)
1311 {
1312         int rc;
1313         ENTRY;
1314
1315         /* Make sure that name isn't exist before doing remote call. */
1316         rc = mdo_lookup(env, md_object_next(mo_p), lname,
1317                         &cmm_env_info(env)->cmi_fid, NULL);
1318         if (rc == 0) {
1319                 rc = -EEXIST;
1320         } else if (rc == -ENOENT) {
1321                 /* Local permission check for name_insert before remote ops. */
1322                 rc = mo_permission(env, NULL, md_object_next(mo_p), NULL,
1323                                    MAY_CREATE);
1324                 if (rc)
1325                         RETURN(rc);
1326
1327                 rc = mo_ref_add(env, md_object_next(mo_s), ma);
1328                 if (rc == 0) {
1329                         ma->ma_attr_flags |= MDS_PERM_BYPASS;
1330                         rc = mdo_name_insert(env, md_object_next(mo_p), lname,
1331                                              lu_object_fid(&mo_s->mo_lu), ma);
1332                         if (unlikely(rc)) {
1333                                 /* TODO: ref_del from mo_s on remote MDS */
1334                                 CWARN("cmr_link failed, should revoke: "
1335                                       "[mo_p "DFID"] [mo_s "DFID"] "
1336                                       "[name %s] [err %d]\n",
1337                                       PFID(lu_object_fid(&mo_p->mo_lu)),
1338                                       PFID(lu_object_fid(&mo_s->mo_lu)),
1339                                       lname->ln_name, rc);
1340                         }
1341                 }
1342         }
1343         RETURN(rc);
1344 }
1345
1346 /**
1347  * Unlink operations for cmr.
1348  *
1349  * The unlink RPC is always issued to the server where parent is living. Hence
1350  * the first operation to do is object unlink on remote server. Second one is
1351  * local mdo_name_remove().
1352  *
1353  * \param mo_p parent md_object. It is local.
1354  * \param mo_c child object to be unlinked. It is remote.
1355  * \param lname Name of file to unlink.
1356  * \param ma object attributes.
1357  */
1358 static int cmr_unlink(const struct lu_env *env, struct md_object *mo_p,
1359                       struct md_object *mo_c, const struct lu_name *lname,
1360                       struct md_attr *ma)
1361 {
1362         struct cmm_thread_info *cmi;
1363         struct md_attr *tmp_ma;
1364         int rc;
1365         ENTRY;
1366
1367         /* Local permission check for name_remove before remote ops. */
1368         rc = mo_permission(env, NULL, md_object_next(mo_p), ma,
1369                            MAY_UNLINK | MAY_VTX_PART);
1370         if (rc)
1371                 RETURN(rc);
1372
1373         /*
1374          * \note \a ma will be changed after mo_ref_del, but we will use
1375          * it for mdo_name_remove() later, so save it before mo_ref_del().
1376          */
1377         cmi = cmm_env_info(env);
1378         tmp_ma = &cmi->cmi_ma;
1379         *tmp_ma = *ma;
1380         rc = mo_ref_del(env, md_object_next(mo_c), ma);
1381         if (rc == 0) {
1382                 tmp_ma->ma_attr_flags |= MDS_PERM_BYPASS;
1383                 rc = mdo_name_remove(env, md_object_next(mo_p), lname, tmp_ma);
1384                 if (unlikely(rc)) {
1385                         /* TODO: ref_add to mo_c on remote MDS */
1386                         CWARN("cmr_unlink failed, should revoke: [mo_p "DFID"]"
1387                               " [mo_c "DFID"] [name %s] [err %d]\n",
1388                               PFID(lu_object_fid(&mo_p->mo_lu)),
1389                               PFID(lu_object_fid(&mo_c->mo_lu)),
1390                               lname->ln_name, rc);
1391                 }
1392         }
1393
1394         RETURN(rc);
1395 }
1396
1397 /** Helper which outputs error message during cmr_rename() */
1398 static inline void cmr_rename_warn(const char *fname,
1399                                   struct md_object *mo_po,
1400                                   struct md_object *mo_pn,
1401                                   const struct lu_fid *lf,
1402                                   const char *s_name,
1403                                   const char *t_name,
1404                                   int err)
1405 {
1406         CWARN("cmr_rename failed for %s, should revoke: "
1407               "[mo_po "DFID"] [mo_pn "DFID"] [lf "DFID"] "
1408               "[sname %s] [tname %s] [err %d]\n", fname,
1409               PFID(lu_object_fid(&mo_po->mo_lu)),
1410               PFID(lu_object_fid(&mo_pn->mo_lu)),
1411               PFID(lf), s_name, t_name, err);
1412 }
1413
1414 /**
1415  * Rename operation for cmr.
1416  *
1417  * This is the most complex cross-reference operation. It may consist of up to 4
1418  * MDS server and require several RPCs to be sent.
1419  *
1420  * \param mo_po Old parent object.
1421  * \param mo_pn New parent object.
1422  * \param lf FID of object to rename.
1423  * \param ls_name Source file name.
1424  * \param mo_t target object. Should be NULL here.
1425  * \param lt_name Name of target file.
1426  * \param ma object attributes.
1427  */
1428 static int cmr_rename(const struct lu_env *env,
1429                       struct md_object *mo_po, struct md_object *mo_pn,
1430                       const struct lu_fid *lf, const struct lu_name *ls_name,
1431                       struct md_object *mo_t, const struct lu_name *lt_name,
1432                       struct md_attr *ma)
1433 {
1434         struct cmm_thread_info *cmi;
1435         struct md_attr *tmp_ma;
1436         int rc;
1437         ENTRY;
1438
1439         LASSERT(mo_t == NULL);
1440
1441         /* get real type of src */
1442         rc = cmm_mode_get(env, md_obj2dev(mo_po), lf, ma, NULL);
1443         if (rc)
1444                 RETURN(rc);
1445
1446         /* Local permission check for name_remove before remote ops. */
1447         rc = mo_permission(env, NULL, md_object_next(mo_po), ma,
1448                            MAY_UNLINK | MAY_VTX_FULL);
1449         if (rc)
1450                 RETURN(rc);
1451
1452         /**
1453          * \todo \a ma maybe changed after mdo_rename_tgt(), but we will use it
1454          * for mdo_name_remove() later, so save it before mdo_rename_tgt.
1455          */
1456         cmi = cmm_env_info(env);
1457         tmp_ma = &cmi->cmi_ma;
1458         *tmp_ma = *ma;
1459         /**
1460          * \note The \a mo_pn is remote directory, so we cannot even know if there is
1461          * \a mo_t or not. Therefore \a mo_t is NULL here but remote server should do
1462          * lookup and process this further.
1463          */
1464         rc = mdo_rename_tgt(env, md_object_next(mo_pn),
1465                             NULL/* mo_t */, lf, lt_name, ma);
1466         if (rc)
1467                 RETURN(rc);
1468
1469         tmp_ma->ma_attr_flags |= MDS_PERM_BYPASS;
1470
1471         /* src object maybe on remote MDS, do remote ops first. */
1472         rc = cmm_rename_ctime(env, md_obj2dev(mo_po), lf, tmp_ma);
1473         if (unlikely(rc)) {
1474                 /* TODO: revoke mdo_rename_tgt */
1475                 cmr_rename_warn("cmm_rename_ctime", mo_po, mo_pn, lf,
1476                                 ls_name->ln_name, lt_name->ln_name, rc);
1477                 RETURN(rc);
1478         }
1479
1480         /* only old name is removed localy */
1481         rc = mdo_name_remove(env, md_object_next(mo_po), ls_name, tmp_ma);
1482         if (unlikely(rc))
1483                 /* TODO: revoke all cmr_rename */
1484                 cmr_rename_warn("mdo_name_remove", mo_po, mo_pn, lf,
1485                                 ls_name->ln_name, lt_name->ln_name, rc);
1486
1487         RETURN(rc);
1488 }
1489
1490 /**
1491  * Part of cross-ref rename().
1492  * Used to insert new name in new parent and unlink target.
1493  */
1494 static int cmr_rename_tgt(const struct lu_env *env,
1495                           struct md_object *mo_p, struct md_object *mo_t,
1496                           const struct lu_fid *lf, const struct lu_name *lname,
1497                           struct md_attr *ma)
1498 {
1499         struct cmm_thread_info *cmi;
1500         struct md_attr *tmp_ma;
1501         int rc;
1502         ENTRY;
1503
1504         /* target object is remote one */
1505         /* Local permission check for rename_tgt before remote ops. */
1506         rc = mo_permission(env, NULL, md_object_next(mo_p), ma,
1507                            MAY_UNLINK | MAY_VTX_PART);
1508         if (rc)
1509                 RETURN(rc);
1510
1511         /*
1512          * XXX: @ma maybe changed after mo_ref_del, but we will use
1513          * it for mdo_rename_tgt later, so save it before mo_ref_del.
1514          */
1515         cmi = cmm_env_info(env);
1516         tmp_ma = &cmi->cmi_ma;
1517         *tmp_ma = *ma;
1518         rc = mo_ref_del(env, md_object_next(mo_t), ma);
1519         /* continue locally with name handling only */
1520         if (rc == 0) {
1521                 tmp_ma->ma_attr_flags |= MDS_PERM_BYPASS;
1522                 rc = mdo_rename_tgt(env, md_object_next(mo_p),
1523                                     NULL, lf, lname, tmp_ma);
1524                 if (unlikely(rc)) {
1525                         /* TODO: ref_add to mo_t on remote MDS */
1526                         CWARN("cmr_rename_tgt failed, should revoke: "
1527                               "[mo_p "DFID"] [mo_t "DFID"] [lf "DFID"] "
1528                               "[name %s] [err %d]\n",
1529                               PFID(lu_object_fid(&mo_p->mo_lu)),
1530                               PFID(lu_object_fid(&mo_t->mo_lu)),
1531                               PFID(lf),
1532                               lname->ln_name, rc);
1533                 }
1534         }
1535         RETURN(rc);
1536 }
1537 /** @} */
1538 /**
1539  * The md_dir_operations for cmr.
1540  */
1541 static const struct md_dir_operations cmr_dir_ops = {
1542         .mdo_is_subdir   = cmm_is_subdir,
1543         .mdo_lookup      = cmr_lookup,
1544         .mdo_lock_mode   = cmr_lock_mode,
1545         .mdo_create      = cmr_create,
1546         .mdo_link        = cmr_link,
1547         .mdo_unlink      = cmr_unlink,
1548         .mdo_lum_lmm_cmp = cmr_lum_lmm_cmp,
1549         .mdo_rename      = cmr_rename,
1550         .mdo_rename_tgt  = cmr_rename_tgt
1551 };
1552 /** @} */