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  *  lustre/cmm/cmm_object.c
   5  *  Lustre Cluster Metadata Manager (cmm)
   6  *
   7  *  Copyright (c) 2006 Cluster File Systems, Inc.
   8  *   Author: Mike Pershin <tappro@clusterfs.com>
   9  *
  10  *   This file is part of the Lustre file system, http://www.lustre.org
  11  *   Lustre is a trademark of Cluster File Systems, Inc.
  12  *
  13  *   You may have signed or agreed to another license before downloading
  14  *   this software.  If so, you are bound by the terms and conditions
  15  *   of that agreement, and the following does not apply to you.  See the
  16  *   LICENSE file included with this distribution for more information.
  17  *
  18  *   If you did not agree to a different license, then this copy of Lustre
  19  *   is open source software; you can redistribute it and/or modify it
  20  *   under the terms of version 2 of the GNU General Public License as
  21  *   published by the Free Software Foundation.
  22  *
  23  *   In either case, Lustre is distributed in the hope that it will be
  24  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  25  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  26  *   license text for more details.
  27  */
  28
  29 #ifndef EXPORT_SYMTAB
  30 # define EXPORT_SYMTAB
  31 #endif
  32
  33 #define DEBUG_SUBSYSTEM S_MDS
  34
  35 #include <lustre_fid.h>
  36 #include "cmm_internal.h"
  37 #include "mdc_internal.h"
  38
  39 extern struct lu_context_key cmm_thread_key;
  40
  41 static int cmm_fld_lookup(struct cmm_device *cm,
  42                           const struct lu_fid *fid, mdsno_t *mds,
  43                           const struct lu_context *ctx)
  44 {
  45         struct lu_site *ls;
  46         int rc = 0;
  47         ENTRY;
  48
  49         LASSERT(fid_is_sane(fid));
  50
  51         ls = cm->cmm_md_dev.md_lu_dev.ld_site;
  52
  53         rc = fld_client_lookup(ls->ls_client_fld,
  54                                fid_seq(fid), mds, ctx);
  55         if (rc) {
  56                 CERROR("can't find mds by seq "LPX64", rc %d\n",
  57                        fid_seq(fid), rc);
  58                 RETURN(rc);
  59         }
  60
  61         if (*mds > cm->cmm_tgt_count) {
  62                 CERROR("Got invalid mdsno: "LPU64" (max: %u)\n",
  63                        *mds, cm->cmm_tgt_count);
  64                 rc = -EINVAL;
  65         } else {
  66                 CDEBUG(D_INFO, "CMM: got MDS "LPU64" for sequence: "LPU64"\n",
  67                        *mds, fid_seq(fid));
  68         }
  69
  70         RETURN (rc);
  71 }
  72
  73 static struct md_object_operations cml_mo_ops;
  74 static struct md_dir_operations    cml_dir_ops;
  75 static struct lu_object_operations cml_obj_ops;
  76
  77 static struct md_object_operations cmr_mo_ops;
  78 static struct md_dir_operations    cmr_dir_ops;
  79 static struct lu_object_operations cmr_obj_ops;
  80
  81 struct lu_object *cmm_object_alloc(const struct lu_context *ctx,
  82                                    const struct lu_object_header *loh,
  83                                    struct lu_device *ld)
  84 {
  85         struct lu_object  *lo = NULL;
  86         const struct lu_fid *fid = &loh->loh_fid;
  87         struct cmm_device *cd;
  88         mdsno_t mdsnum;
  89         int rc = 0;
  90
  91         ENTRY;
  92
  93         cd = lu2cmm_dev(ld);
  94         if (cd->cmm_flags & CMM_INITIALIZED) {
  95                 /* get object location */
  96                 rc = cmm_fld_lookup(lu2cmm_dev(ld), fid, &mdsnum, ctx);
  97                 if (rc)
  98                         RETURN(NULL);
  99         } else
 100                 /*
 101                  * Device is not yet initialized, cmm_object is being created
 102                  * as part of early bootstrap procedure (it is /ROOT, or /fld,
 103                  * etc.). Such object *has* to be local.
 104                  */
 105                 mdsnum = cd->cmm_local_num;
 106
 107         /* select the proper set of operations based on object location */
 108         if (mdsnum == cd->cmm_local_num) {
 109                 struct cml_object *clo;
 110
 111                 OBD_ALLOC_PTR(clo);
 112                 if (clo != NULL) {
 113                         lo = &clo->cmm_obj.cmo_obj.mo_lu;
 114                         lu_object_init(lo, NULL, ld);
 115                         clo->cmm_obj.cmo_obj.mo_ops = &cml_mo_ops;
 116                         clo->cmm_obj.cmo_obj.mo_dir_ops = &cml_dir_ops;
 117                         lo->lo_ops = &cml_obj_ops;
 118                 }
 119         } else {
 120                 struct cmr_object *cro;
 121
 122                 OBD_ALLOC_PTR(cro);
 123                 if (cro != NULL) {
 124                         lo = &cro->cmm_obj.cmo_obj.mo_lu;
 125                         lu_object_init(lo, NULL, ld);
 126                         cro->cmm_obj.cmo_obj.mo_ops = &cmr_mo_ops;
 127                         cro->cmm_obj.cmo_obj.mo_dir_ops = &cmr_dir_ops;
 128                         lo->lo_ops = &cmr_obj_ops;
 129                         cro->cmo_num = mdsnum;
 130                 }
 131         }
 132         RETURN(lo);
 133 }
 134
 135 /*
 136  * CMM has two types of objects - local and remote. They have different set
 137  * of operations so we are avoiding multiple checks in code.
 138  */
 139
 140 /*
 141  * local CMM object operations. cml_...
 142  */
 143 static inline struct cml_object *lu2cml_obj(struct lu_object *o)
 144 {
 145         return container_of0(o, struct cml_object, cmm_obj.cmo_obj.mo_lu);
 146 }
 147 static inline struct cml_object *md2cml_obj(struct md_object *mo)
 148 {
 149         return container_of0(mo, struct cml_object, cmm_obj.cmo_obj);
 150 }
 151 static inline struct cml_object *cmm2cml_obj(struct cmm_object *co)
 152 {
 153         return container_of0(co, struct cml_object, cmm_obj);
 154 }
 155 /* get local child device */
 156 static struct lu_device *cml_child_dev(struct cmm_device *d)
 157 {
 158         return &d->cmm_child->md_lu_dev;
 159 }
 160
 161 /* lu_object operations */
 162 static void cml_object_free(const struct lu_context *ctx,
 163                             struct lu_object *lo)
 164 {
 165         struct cml_object *clo = lu2cml_obj(lo);
 166         lu_object_fini(lo);
 167         OBD_FREE_PTR(clo);
 168 }
 169
 170 static int cml_object_init(const struct lu_context *ctx, struct lu_object *lo)
 171 {
 172         struct cmm_device *cd = lu2cmm_dev(lo->lo_dev);
 173         struct lu_device  *c_dev;
 174         struct lu_object  *c_obj;
 175         int rc;
 176
 177         ENTRY;
 178
 179         c_dev = cml_child_dev(cd);
 180         if (c_dev == NULL) {
 181                 rc = -ENOENT;
 182         } else {
 183                 c_obj = c_dev->ld_ops->ldo_object_alloc(ctx,
 184                                                         lo->lo_header, c_dev);
 185                 if (c_obj != NULL) {
 186                         lu_object_add(lo, c_obj);
 187                         rc = 0;
 188                 } else {
 189                         rc = -ENOMEM;
 190                 }
 191         }
 192
 193         RETURN(rc);
 194 }
 195
 196 static int cml_object_print(const struct lu_context *ctx, void *cookie,
 197                             lu_printer_t p, const struct lu_object *lo)
 198 {
 199         return (*p)(ctx, cookie, LUSTRE_CMM_NAME"-local@%p", lo);
 200 }
 201
 202 static struct lu_object_operations cml_obj_ops = {
 203         .loo_object_init    = cml_object_init,
 204         .loo_object_free    = cml_object_free,
 205         .loo_object_print   = cml_object_print
 206 };
 207
 208 /* CMM local md_object operations */
 209 static int cml_object_create(const struct lu_context *ctx,
 210                              struct md_object *mo,
 211                              const struct md_create_spec *spec,
 212                              struct md_attr *attr,
 213                              struct md_ucred *uc)
 214 {
 215         int rc;
 216         ENTRY;
 217         rc = mo_object_create(ctx, md_object_next(mo), spec, attr, uc);
 218         RETURN(rc);
 219 }
 220
 221 static int cml_permission(const struct lu_context *ctx,
 222                         struct md_object *mo, int mask, struct md_ucred *uc)
 223 {
 224         int rc;
 225         ENTRY;
 226         rc = mo_permission(ctx, md_object_next(mo), mask, uc);
 227         RETURN(rc);
 228 }
 229
 230 static int cml_attr_get(const struct lu_context *ctx, struct md_object *mo,
 231                         struct md_attr *attr, struct md_ucred *uc)
 232 {
 233         int rc;
 234         ENTRY;
 235         rc = mo_attr_get(ctx, md_object_next(mo), attr, uc);
 236         RETURN(rc);
 237 }
 238
 239 static int cml_attr_set(const struct lu_context *ctx, struct md_object *mo,
 240                         const struct md_attr *attr, struct md_ucred *uc)
 241 {
 242         int rc;
 243         ENTRY;
 244         rc = mo_attr_set(ctx, md_object_next(mo), attr, uc);
 245         RETURN(rc);
 246 }
 247
 248 static int cml_xattr_get(const struct lu_context *ctx, struct md_object *mo,
 249                          void *buf, int buflen, const char *name,
 250                          struct md_ucred *uc)
 251 {
 252         int rc;
 253         ENTRY;
 254         rc = mo_xattr_get(ctx, md_object_next(mo), buf, buflen, name, uc);
 255         RETURN(rc);
 256 }
 257
 258 static int cml_readlink(const struct lu_context *ctx, struct md_object *mo,
 259                         void *buf, int buflen, struct md_ucred *uc)
 260 {
 261         int rc;
 262         ENTRY;
 263         rc = mo_readlink(ctx, md_object_next(mo), buf, buflen, uc);
 264         RETURN(rc);
 265 }
 266
 267 static int cml_xattr_list(const struct lu_context *ctx, struct md_object *mo,
 268                           void *buf, int buflen, struct md_ucred *uc)
 269 {
 270         int rc;
 271         ENTRY;
 272         rc = mo_xattr_list(ctx, md_object_next(mo), buf, buflen, uc);
 273         RETURN(rc);
 274 }
 275
 276 static int cml_xattr_set(const struct lu_context *ctx, struct md_object *mo,
 277                          const void *buf, int buflen,
 278                          const char *name, int fl, struct md_ucred *uc)
 279 {
 280         int rc;
 281         ENTRY;
 282         rc = mo_xattr_set(ctx, md_object_next(mo), buf, buflen, name, fl, uc);
 283         RETURN(rc);
 284 }
 285
 286 static int cml_xattr_del(const struct lu_context *ctx, struct md_object *mo,
 287                          const char *name, struct md_ucred *uc)
 288 {
 289         int rc;
 290         ENTRY;
 291         rc = mo_xattr_del(ctx, md_object_next(mo), name, uc);
 292         RETURN(rc);
 293 }
 294
 295 static int cml_ref_add(const struct lu_context *ctx, struct md_object *mo,
 296                        struct md_ucred *uc)
 297 {
 298         int rc;
 299         ENTRY;
 300         rc = mo_ref_add(ctx, md_object_next(mo), uc);
 301         RETURN(rc);
 302 }
 303
 304 static int cml_ref_del(const struct lu_context *ctx, struct md_object *mo,
 305                        struct md_attr *ma, struct md_ucred *uc)
 306 {
 307         int rc;
 308         ENTRY;
 309         rc = mo_ref_del(ctx, md_object_next(mo), ma, uc);
 310         RETURN(rc);
 311 }
 312
 313 static int cml_open(const struct lu_context *ctx, struct md_object *mo,
 314                     int flags, struct md_ucred *uc)
 315 {
 316         int rc;
 317         ENTRY;
 318         rc = mo_open(ctx, md_object_next(mo), flags, uc);
 319         RETURN(rc);
 320 }
 321
 322 static int cml_close(const struct lu_context *ctx, struct md_object *mo,
 323                      struct md_attr *ma, struct md_ucred *uc)
 324 {
 325         int rc;
 326         ENTRY;
 327         rc = mo_close(ctx, md_object_next(mo), ma, uc);
 328         RETURN(rc);
 329 }
 330
 331 static int cml_readpage(const struct lu_context *ctxt, struct md_object *mo,
 332                         const struct lu_rdpg *rdpg, struct md_ucred *uc)
 333 {
 334         int rc;
 335         ENTRY;
 336         rc = mo_readpage(ctxt, md_object_next(mo), rdpg, uc);
 337         RETURN(rc);
 338 }
 339
 340 static int cml_capa_get(const struct lu_context *ctxt, struct md_object *mo,
 341                         struct lustre_capa *capa)
 342 {
 343         int rc;
 344         ENTRY;
 345         rc = mo_capa_get(ctxt, md_object_next(mo), capa);
 346         RETURN(rc);
 347 }
 348
 349 static struct md_object_operations cml_mo_ops = {
 350         .moo_permission    = cml_permission,
 351         .moo_attr_get      = cml_attr_get,
 352         .moo_attr_set      = cml_attr_set,
 353         .moo_xattr_get     = cml_xattr_get,
 354         .moo_xattr_list    = cml_xattr_list,
 355         .moo_xattr_set     = cml_xattr_set,
 356         .moo_xattr_del     = cml_xattr_del,
 357         .moo_object_create = cml_object_create,
 358         .moo_ref_add       = cml_ref_add,
 359         .moo_ref_del       = cml_ref_del,
 360         .moo_open          = cml_open,
 361         .moo_close         = cml_close,
 362         .moo_readpage      = cml_readpage,
 363         .moo_readlink      = cml_readlink,
 364         .moo_capa_get      = cml_capa_get
 365 };
 366
 367 /* md_dir operations */
 368 static int cml_lookup(const struct lu_context *ctx, struct md_object *mo_p,
 369                       const char *name, struct lu_fid *lf, struct md_ucred *uc)
 370 {
 371         int rc;
 372         ENTRY;
 373         rc = mdo_lookup(ctx, md_object_next(mo_p), name, lf, uc);
 374         RETURN(rc);
 375
 376 }
 377
 378 static int cml_create(const struct lu_context *ctx,
 379                       struct md_object *mo_p, const char *child_name,
 380                       struct md_object *mo_c, const struct md_create_spec *spec,
 381                       struct md_attr *ma, struct md_ucred *uc)
 382 {
 383         int rc;
 384         ENTRY;
 385
 386 #ifdef HAVE_SPLIT_SUPPORT
 387         rc = cml_try_to_split(ctx, mo_p, uc);
 388         if (rc)
 389                 RETURN(rc);
 390 #endif
 391
 392         rc = mdo_create(ctx, md_object_next(mo_p), child_name,
 393                         md_object_next(mo_c), spec, ma, uc);
 394
 395
 396         RETURN(rc);
 397 }
 398
 399 static int cml_create_data(const struct lu_context *ctx, struct md_object *p,
 400                            struct md_object *o,
 401                            const struct md_create_spec *spec,
 402                            struct md_attr *ma, struct md_ucred *uc)
 403 {
 404         int rc;
 405         ENTRY;
 406         rc = mdo_create_data(ctx, md_object_next(p), md_object_next(o),
 407                              spec, ma, uc);
 408         RETURN(rc);
 409 }
 410
 411 static int cml_link(const struct lu_context *ctx, struct md_object *mo_p,
 412                     struct md_object *mo_s, const char *name,
 413                     struct md_attr *ma, struct md_ucred *uc)
 414 {
 415         int rc;
 416         ENTRY;
 417         rc = mdo_link(ctx, md_object_next(mo_p), md_object_next(mo_s),
 418                       name, ma, uc);
 419         RETURN(rc);
 420 }
 421
 422 static int cml_unlink(const struct lu_context *ctx, struct md_object *mo_p,
 423                       struct md_object *mo_c, const char *name,
 424                       struct md_attr *ma, struct md_ucred *uc)
 425 {
 426         int rc;
 427         ENTRY;
 428         rc = mdo_unlink(ctx, md_object_next(mo_p), md_object_next(mo_c),
 429                         name, ma, uc);
 430         RETURN(rc);
 431 }
 432
 433 /* rename is split to local/remote by location of new parent dir */
 434 struct md_object *md_object_find(const struct lu_context *ctx,
 435                                  struct md_device *md,
 436                                  const struct lu_fid *f)
 437 {
 438         struct lu_object *o;
 439         struct md_object *m;
 440         ENTRY;
 441
 442         o = lu_object_find(ctx, md2lu_dev(md)->ld_site, f, BYPASS_CAPA);
 443         if (IS_ERR(o))
 444                 m = (struct md_object *)o;
 445         else {
 446                 o = lu_object_locate(o->lo_header, md2lu_dev(md)->ld_type);
 447                 m = o ? lu2md(o) : NULL;
 448         }
 449         RETURN(m);
 450 }
 451
 452 static int __cmm_mode_get(const struct lu_context *ctx, struct md_device *md,
 453                           const struct lu_fid *lf, struct md_attr *ma,
 454                           struct md_ucred *uc)
 455 {
 456         struct cmm_thread_info *cmi;
 457         struct md_object *mo_s = md_object_find(ctx, md, lf);
 458         struct md_attr *tmp_ma;
 459         int rc;
 460         ENTRY;
 461
 462         if (IS_ERR(mo_s))
 463                 RETURN(PTR_ERR(mo_s));
 464
 465         cmi = lu_context_key_get(ctx, &cmm_thread_key);
 466         LASSERT(cmi);
 467         tmp_ma = &cmi->cmi_ma;
 468         tmp_ma->ma_need = MA_INODE;
 469
 470         /* get type from src, can be remote req */
 471         rc = mo_attr_get(ctx, md_object_next(mo_s), tmp_ma, uc);
 472         if (rc == 0) {
 473                 ma->ma_attr.la_mode = tmp_ma->ma_attr.la_mode;
 474                 ma->ma_attr.la_flags = tmp_ma->ma_attr.la_flags;
 475                 ma->ma_attr.la_valid |= LA_MODE | LA_FLAGS;
 476         }
 477         lu_object_put(ctx, &mo_s->mo_lu);
 478         return rc;
 479 }
 480
 481 static int cml_rename(const struct lu_context *ctx, struct md_object *mo_po,
 482                       struct md_object *mo_pn, const struct lu_fid *lf,
 483                       const char *s_name, struct md_object *mo_t,
 484                       const char *t_name, struct md_attr *ma,
 485                       struct md_ucred *uc)
 486 {
 487         int rc;
 488         ENTRY;
 489
 490         rc = __cmm_mode_get(ctx, md_obj2dev(mo_po), lf, ma, uc);
 491         if (rc != 0)
 492                 RETURN(rc);
 493
 494         if (mo_t && lu_object_exists(&mo_t->mo_lu) < 0) {
 495                 /* mo_t is remote object and there is RPC to unlink it */
 496                 rc = mo_ref_del(ctx, md_object_next(mo_t), ma, uc);
 497                 if (rc)
 498                         RETURN(rc);
 499                 mo_t = NULL;
 500         }
 501
 502         /* local rename, mo_t can be NULL */
 503         rc = mdo_rename(ctx, md_object_next(mo_po),
 504                         md_object_next(mo_pn), lf, s_name,
 505                         md_object_next(mo_t), t_name, ma, uc);
 506         RETURN(rc);
 507 }
 508
 509 static int cml_rename_tgt(const struct lu_context *ctx, struct md_object *mo_p,
 510                           struct md_object *mo_t, const struct lu_fid *lf,
 511                           const char *name, struct md_attr *ma,
 512                           struct md_ucred *uc)
 513 {
 514         int rc;
 515         ENTRY;
 516
 517         rc = mdo_rename_tgt(ctx, md_object_next(mo_p),
 518                             md_object_next(mo_t), lf, name, ma, uc);
 519         RETURN(rc);
 520 }
 521 /* used only in case of rename_tgt() when target is not exist */
 522 static int cml_name_insert(const struct lu_context *ctx, struct md_object *p,
 523                            const char *name, const struct lu_fid *lf, int isdir,
 524                            struct md_ucred *uc)
 525 {
 526         int rc;
 527         ENTRY;
 528
 529         rc = mdo_name_insert(ctx, md_object_next(p), name, lf, isdir, uc);
 530
 531         RETURN(rc);
 532 }
 533
 534 /* Common method for remote and local use. */
 535 static int cmm_is_subdir(const struct lu_context *ctx, struct md_object *mo,
 536                          const struct lu_fid *fid, struct lu_fid *sfid,
 537                          struct md_ucred *uc)
 538 {
 539         struct cmm_thread_info *cmi;
 540         int rc;
 541         ENTRY;
 542
 543         cmi = lu_context_key_get(ctx, &cmm_thread_key);
 544         rc = __cmm_mode_get(ctx, md_obj2dev(mo), fid, &cmi->cmi_ma, uc);
 545         if (rc)
 546                 RETURN(rc);
 547
 548         if (!S_ISDIR(cmi->cmi_ma.ma_attr.la_mode))
 549                 RETURN(0);
 550
 551         rc = mdo_is_subdir(ctx, md_object_next(mo), fid, sfid, uc);
 552         RETURN(rc);
 553 }
 554
 555 static struct md_dir_operations cml_dir_ops = {
 556         .mdo_is_subdir   = cmm_is_subdir,
 557         .mdo_lookup      = cml_lookup,
 558         .mdo_create      = cml_create,
 559         .mdo_link        = cml_link,
 560         .mdo_unlink      = cml_unlink,
 561         .mdo_name_insert = cml_name_insert,
 562         .mdo_rename      = cml_rename,
 563         .mdo_rename_tgt  = cml_rename_tgt,
 564         .mdo_create_data = cml_create_data
 565 };
 566
 567 /* -------------------------------------------------------------------
 568  * remote CMM object operations. cmr_...
 569  */
 570 static inline struct cmr_object *lu2cmr_obj(struct lu_object *o)
 571 {
 572         return container_of0(o, struct cmr_object, cmm_obj.cmo_obj.mo_lu);
 573 }
 574 static inline struct cmr_object *md2cmr_obj(struct md_object *mo)
 575 {
 576         return container_of0(mo, struct cmr_object, cmm_obj.cmo_obj);
 577 }
 578 static inline struct cmr_object *cmm2cmr_obj(struct cmm_object *co)
 579 {
 580         return container_of0(co, struct cmr_object, cmm_obj);
 581 }
 582
 583 /* get proper child device from MDCs */
 584 static struct lu_device *cmr_child_dev(struct cmm_device *d, __u32 num)
 585 {
 586         struct lu_device *next = NULL;
 587         struct mdc_device *mdc;
 588
 589         spin_lock(&d->cmm_tgt_guard);
 590         list_for_each_entry(mdc, &d->cmm_targets, mc_linkage) {
 591                 if (mdc->mc_num == num) {
 592                         next = mdc2lu_dev(mdc);
 593                         break;
 594                 }
 595         }
 596         spin_unlock(&d->cmm_tgt_guard);
 597         return next;
 598 }
 599
 600 /* lu_object operations */
 601 static void cmr_object_free(const struct lu_context *ctx,
 602                             struct lu_object *lo)
 603 {
 604         struct cmr_object *cro = lu2cmr_obj(lo);
 605         lu_object_fini(lo);
 606         OBD_FREE_PTR(cro);
 607 }
 608
 609 static int cmr_object_init(const struct lu_context *ctx, struct lu_object *lo)
 610 {
 611         struct cmm_device *cd = lu2cmm_dev(lo->lo_dev);
 612         struct lu_device  *c_dev;
 613         struct lu_object  *c_obj;
 614         int rc;
 615
 616         ENTRY;
 617
 618         c_dev = cmr_child_dev(cd, lu2cmr_obj(lo)->cmo_num);
 619         if (c_dev == NULL) {
 620                 rc = -ENOENT;
 621         } else {
 622                 c_obj = c_dev->ld_ops->ldo_object_alloc(ctx,
 623                                                         lo->lo_header, c_dev);
 624                 if (c_obj != NULL) {
 625                         lu_object_add(lo, c_obj);
 626                         rc = 0;
 627                 } else {
 628                         rc = -ENOMEM;
 629                 }
 630         }
 631
 632         RETURN(rc);
 633 }
 634
 635 static int cmr_object_print(const struct lu_context *ctx, void *cookie,
 636                             lu_printer_t p, const struct lu_object *lo)
 637 {
 638         return (*p)(ctx, cookie, LUSTRE_CMM_NAME"-remote@%p", lo);
 639 }
 640
 641 static struct lu_object_operations cmr_obj_ops = {
 642         .loo_object_init    = cmr_object_init,
 643         .loo_object_free    = cmr_object_free,
 644         .loo_object_print   = cmr_object_print
 645 };
 646
 647 /* CMM remote md_object operations. All are invalid */
 648 static int cmr_object_create(const struct lu_context *ctx,
 649                              struct md_object *mo,
 650                              const struct md_create_spec *spec,
 651                              struct md_attr *ma,
 652                              struct md_ucred *uc)
 653 {
 654         RETURN(-EFAULT);
 655 }
 656
 657 static int cmr_permission(const struct lu_context *ctx, struct md_object *mo,
 658                           int mask, struct md_ucred *uc)
 659 {
 660         RETURN(-EREMOTE);
 661 }
 662
 663 static int cmr_attr_get(const struct lu_context *ctx, struct md_object *mo,
 664                         struct md_attr *attr, struct md_ucred *uc)
 665 {
 666         RETURN(-EREMOTE);
 667 }
 668
 669 static int cmr_attr_set(const struct lu_context *ctx, struct md_object *mo,
 670                         const struct md_attr *attr, struct md_ucred *uc)
 671 {
 672         RETURN(-EFAULT);
 673 }
 674
 675 static int cmr_xattr_get(const struct lu_context *ctx, struct md_object *mo,
 676                          void *buf, int buflen, const char *name,
 677                          struct md_ucred *uc)
 678 {
 679         RETURN(-EFAULT);
 680 }
 681
 682 static int cmr_readlink(const struct lu_context *ctx, struct md_object *mo,
 683                         void *buf, int buflen, struct md_ucred *uc)
 684 {
 685         RETURN(-EFAULT);
 686 }
 687
 688 static int cmr_xattr_list(const struct lu_context *ctx, struct md_object *mo,
 689                           void *buf, int buflen, struct md_ucred *uc)
 690 {
 691         RETURN(-EFAULT);
 692 }
 693
 694 static int cmr_xattr_set(const struct lu_context *ctx, struct md_object *mo,
 695                          const void *buf, int buflen, const char *name, int fl,
 696                          struct md_ucred *uc)
 697 {
 698         RETURN(-EFAULT);
 699 }
 700
 701 static int cmr_xattr_del(const struct lu_context *ctx, struct md_object *mo,
 702                          const char *name, struct md_ucred *uc)
 703 {
 704         RETURN(-EFAULT);
 705 }
 706
 707 static int cmr_ref_add(const struct lu_context *ctx, struct md_object *mo,
 708                        struct md_ucred *uc)
 709 {
 710         RETURN(-EFAULT);
 711 }
 712
 713 static int cmr_ref_del(const struct lu_context *ctx, struct md_object *mo,
 714                        struct md_attr *ma, struct md_ucred *uc)
 715 {
 716         RETURN(-EFAULT);
 717 }
 718
 719 static int cmr_open(const struct lu_context *ctx, struct md_object *mo,
 720                     int flags, struct md_ucred *uc)
 721 {
 722         RETURN(-EREMOTE);
 723 }
 724
 725 static int cmr_close(const struct lu_context *ctx, struct md_object *mo,
 726                      struct md_attr *ma, struct md_ucred *uc)
 727 {
 728         RETURN(-EFAULT);
 729 }
 730
 731 static int cmr_readpage(const struct lu_context *ctxt, struct md_object *mo,
 732                         const struct lu_rdpg *rdpg, struct md_ucred *uc)
 733 {
 734         RETURN(-EREMOTE);
 735 }
 736
 737 static int cmr_capa_get(const struct lu_context *ctxt, struct md_object *mo,
 738                         struct lustre_capa *capa)
 739 {
 740         RETURN(-EFAULT);
 741 }
 742
 743 static struct md_object_operations cmr_mo_ops = {
 744         .moo_permission    = cmr_permission,
 745         .moo_attr_get      = cmr_attr_get,
 746         .moo_attr_set      = cmr_attr_set,
 747         .moo_xattr_get     = cmr_xattr_get,
 748         .moo_xattr_set     = cmr_xattr_set,
 749         .moo_xattr_list    = cmr_xattr_list,
 750         .moo_xattr_del     = cmr_xattr_del,
 751         .moo_object_create = cmr_object_create,
 752         .moo_ref_add       = cmr_ref_add,
 753         .moo_ref_del       = cmr_ref_del,
 754         .moo_open          = cmr_open,
 755         .moo_close         = cmr_close,
 756         .moo_readpage      = cmr_readpage,
 757         .moo_readlink      = cmr_readlink,
 758         .moo_capa_get      = cmr_capa_get
 759 };
 760
 761 /* remote part of md_dir operations */
 762 static int cmr_lookup(const struct lu_context *ctx, struct md_object *mo_p,
 763                       const char *name, struct lu_fid *lf, struct md_ucred *uc)
 764 {
 765         /*
 766          * This can happens while rename() If new parent is remote dir, lookup
 767          * will happen here.
 768          */
 769
 770         RETURN(-EREMOTE);
 771 }
 772
 773 /*
 774  * All methods below are cross-ref by nature. They consist of remote call and
 775  * local operation. Due to future rollback functionality there are several
 776  * limitations for such methods:
 777  * 1) remote call should be done at first to do epoch negotiation between all
 778  * MDS involved and to avoid the RPC inside transaction.
 779  * 2) only one RPC can be sent - also due to epoch negotiation.
 780  * For more details see rollback HLD/DLD.
 781  *
 782  */
 783 static int cmr_create(const struct lu_context *ctx, struct md_object *mo_p,
 784                       const char *child_name, struct md_object *mo_c,
 785                       const struct md_create_spec *spec,
 786                       struct md_attr *ma, struct md_ucred *uc)
 787 {
 788         struct cmm_thread_info *cmi;
 789         struct md_attr *tmp_ma;
 790         int rc;
 791
 792         ENTRY;
 793         /* check the SGID attr */
 794         cmi = lu_context_key_get(ctx, &cmm_thread_key);
 795         LASSERT(cmi);
 796         tmp_ma = &cmi->cmi_ma;
 797         tmp_ma->ma_need = MA_INODE;
 798         rc = mo_attr_get(ctx, md_object_next(mo_p), tmp_ma, uc);
 799         if (rc)
 800                 RETURN(rc);
 801
 802         if (tmp_ma->ma_attr.la_mode & S_ISGID) {
 803                 ma->ma_attr.la_gid = tmp_ma->ma_attr.la_gid;
 804                 if (S_ISDIR(ma->ma_attr.la_mode)) {
 805                         ma->ma_attr.la_mode |= S_ISGID;
 806                         ma->ma_attr.la_valid |= LA_MODE;
 807                 }
 808         }
 809         /* remote object creation and local name insert */
 810         rc = mo_object_create(ctx, md_object_next(mo_c), spec, ma, uc);
 811         if (rc == 0) {
 812                 rc = mdo_name_insert(ctx, md_object_next(mo_p),
 813                                      child_name, lu_object_fid(&mo_c->mo_lu),
 814                                      S_ISDIR(ma->ma_attr.la_mode), uc);
 815         }
 816
 817         RETURN(rc);
 818 }
 819
 820 static int cmr_link(const struct lu_context *ctx, struct md_object *mo_p,
 821                     struct md_object *mo_s, const char *name,
 822                     struct md_attr *ma, struct md_ucred *uc)
 823 {
 824         int rc;
 825         ENTRY;
 826
 827         //XXX: make sure that MDT checks name isn't exist
 828
 829         rc = mo_ref_add(ctx, md_object_next(mo_s), uc);
 830         if (rc == 0) {
 831                 rc = mdo_name_insert(ctx, md_object_next(mo_p),
 832                                      name, lu_object_fid(&mo_s->mo_lu), 0, uc);
 833         }
 834
 835         RETURN(rc);
 836 }
 837
 838 static int cmr_unlink(const struct lu_context *ctx, struct md_object *mo_p,
 839                       struct md_object *mo_c, const char *name,
 840                       struct md_attr *ma, struct md_ucred *uc)
 841 {
 842         int rc;
 843         ENTRY;
 844
 845         rc = mo_ref_del(ctx, md_object_next(mo_c), ma, uc);
 846         if (rc == 0) {
 847                 rc = mdo_name_remove(ctx, md_object_next(mo_p), name, uc);
 848         }
 849
 850         RETURN(rc);
 851 }
 852
 853 static int cmr_rename(const struct lu_context *ctx,
 854                       struct md_object *mo_po, struct md_object *mo_pn,
 855                       const struct lu_fid *lf, const char *s_name,
 856                       struct md_object *mo_t, const char *t_name,
 857                       struct md_attr *ma, struct md_ucred *uc)
 858 {
 859         int rc;
 860         ENTRY;
 861
 862         /* get real type of src */
 863         rc = __cmm_mode_get(ctx, md_obj2dev(mo_po), lf, ma, uc);
 864         if (rc != 0)
 865                 RETURN(rc);
 866
 867         LASSERT(mo_t == NULL);
 868         /* the mo_pn is remote directory, so we cannot even know if there is
 869          * mo_t or not. Therefore mo_t is NULL here but remote server should do
 870          * lookup and process this further */
 871         rc = mdo_rename_tgt(ctx, md_object_next(mo_pn),
 872                             NULL/* mo_t */, lf, t_name, ma, uc);
 873         /* only old name is removed localy */
 874         if (rc == 0)
 875                 rc = mdo_name_remove(ctx, md_object_next(mo_po),
 876                                      s_name, uc);
 877
 878         RETURN(rc);
 879 }
 880
 881 /* part of cross-ref rename(). Used to insert new name in new parent
 882  * and unlink target with same name if it exists */
 883 static int cmr_rename_tgt(const struct lu_context *ctx,
 884                           struct md_object *mo_p, struct md_object *mo_t,
 885                           const struct lu_fid *lf, const char *name,
 886                           struct md_attr *ma, struct md_ucred *uc)
 887 {
 888         int rc;
 889         ENTRY;
 890         /* target object is remote one */
 891         rc = mo_ref_del(ctx, md_object_next(mo_t), ma, uc);
 892         /* continue locally with name handling only */
 893         if (rc == 0)
 894                 rc = mdo_rename_tgt(ctx, md_object_next(mo_p),
 895                                     NULL, lf, name, ma, uc);
 896         RETURN(rc);
 897 }
 898
 899 static struct md_dir_operations cmr_dir_ops = {
 900         .mdo_is_subdir   = cmm_is_subdir,
 901         .mdo_lookup      = cmr_lookup,
 902         .mdo_create      = cmr_create,
 903         .mdo_link        = cmr_link,
 904         .mdo_unlink      = cmr_unlink,
 905         .mdo_rename      = cmr_rename,
 906         .mdo_rename_tgt  = cmr_rename_tgt,
 907 };