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