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 "cmm_internal.h"
  36 #include "mdc_internal.h"
  37
  38 #ifdef CMM_CODE
  39 static int cmm_fld_lookup(const struct lu_fid *fid)
  40 {
  41         int rc;
  42         /* temporary hack for proto mkdir */
  43         rc = (unsigned long)fid_seq(fid) / LUSTRE_SEQ_RANGE;
  44         CWARN("Get MDS %d for sequence: "LPU64"\n", rc, fid_seq(fid));
  45         RETURN(rc);
  46 }
  47
  48 static struct md_object_operations cml_mo_ops;
  49 static struct md_dir_operations    cml_dir_ops;
  50 static struct lu_object_operations cml_obj_ops;
  51
  52 static struct md_object_operations cmr_mo_ops;
  53 static struct md_dir_operations    cmr_dir_ops;
  54 static struct lu_object_operations cmr_obj_ops;
  55
  56 struct lu_object *cmm_object_alloc(const struct lu_context *ctx,
  57                                    const struct lu_object_header *loh,
  58                                    struct lu_device *ld)
  59 {
  60         struct lu_object  *lo = NULL;
  61         const struct lu_fid *fid = loh->loh_fid;
  62         int mdsnum, rc;
  63         ENTRY;
  64
  65         /* get object location */
  66         mdsnum = cmm_fld_lookup(fid);
  67
  68         /* select the proper set of operations based on object location */
  69         if (mdsnum == lu2cmm_dev(ld)->cmm_local_num) {
  70                 struct cml_object *clo;
  71
  72                 OBD_ALLOC_PTR(clo);
  73                 if (clo != NULL) {
  74                         lo = &clo->cmm_obj.cmo_obj.mo_lu;
  75                         lu_object_init(lo, NULL, ld);
  76                         clo->cmm_obj.cmo_obj.mo_ops = &cml_mo_ops;
  77                         clo->cmm_obj.cmo_obj.mo_dir_ops = &cml_dir_ops;
  78                         lo->lo_ops = &cml_obj_ops;
  79                 }
  80         } else {
  81                 struct cmr_object *cro;
  82
  83                 OBD_ALLOC_PTR(cro);
  84                 if (cro != NULL) {
  85                         lo = &cro->cmm_obj.cmo_obj.mo_lu;
  86                         lu_object_init(lo, NULL, ld);
  87                         cro->cmm_obj.cmo_obj.mo_ops = &cmr_mo_ops;
  88                         cro->cmm_obj.cmo_obj.mo_dir_ops = &cmr_dir_ops;
  89                         lo->lo_ops = &cmr_obj_ops;
  90                         cro->cmo_num = mdsnum;
  91                 }
  92         }
  93         RETURN(lo);
  94 }
  95
  96 /*
  97  * CMM has two types of objects - local and remote. They have different set
  98  * of operations so we are avoiding multiple checks in code.
  99  */
 100
 101 /*
 102  * local CMM object operations. cml_...
 103  */
 104 static inline struct cml_object *lu2cml_obj(struct lu_object *o)
 105 {
 106         return container_of0(o, struct cml_object, cmm_obj.cmo_obj.mo_lu);
 107 }
 108 static inline struct cml_object *md2cml_obj(struct md_object *mo)
 109 {
 110         return container_of0(mo, struct cml_object, cmm_obj.cmo_obj);
 111 }
 112 static inline struct cml_object *cmm2cml_obj(struct cmm_object *co)
 113 {
 114         return container_of0(co, struct cml_object, cmm_obj);
 115 }
 116 /* get local child device */
 117 static struct lu_device *cml_child_dev(struct cmm_device *d)
 118 {
 119         return next = &d->cmm_child->md_lu_dev;
 120 }
 121
 122 /* lu_object operations */
 123 static void cml_object_free(const struct lu_context *ctx,
 124                             struct lu_object *lo)
 125 {
 126         struct cml_object *clo = lu2cml_obj(lo);
 127         lu_object_fini(lo);
 128         OBD_FREE_PTR(clo);
 129 }
 130
 131 static int cml_object_init(const struct lu_context *ctx, struct lu_object *lo)
 132 {
 133         struct cmm_device *cd = lu2cmm_dev(lo->lo_dev);
 134         struct lu_device  *c_dev;
 135         struct lu_object  *c_obj;
 136         int rc;
 137
 138         ENTRY;
 139
 140         c_dev = cml_child_dev(cd);
 141         if (c_dev == NULL) {
 142                 rc = -ENOENT;
 143         } else {
 144                 c_obj = c_dev->ld_ops->ldo_object_alloc(ctx,
 145                                                         lo->lo_header, c_dev);
 146                 if (c_obj != NULL) {
 147                         lu_object_add(lo, c_obj);
 148                         rc = 0;
 149                 } else {
 150                         rc = -ENOMEM;
 151                 }
 152         }
 153
 154         RETURN(rc);
 155 }
 156
 157 static int cml_object_exists(const struct lu_context *ctx,
 158                              struct lu_object *lo)
 159 {
 160         return lu_object_exists(ctx, lu_object_next(lo));
 161 }
 162
 163 static int cml_object_print(const struct lu_context *ctx,
 164                             struct seq_file *f, const struct lu_object *lo)
 165 {
 166         return seq_printf(f, LUSTRE_CMM0_NAME"-object@%p", lo);
 167 }
 168
 169 static struct lu_object_operations cml_obj_ops = {
 170         .loo_object_init    = cml_object_init,
 171         .loo_object_free    = cml_object_free,
 172         .loo_object_print   = cml_object_print,
 173         .loo_object_exists  = cml_object_exists
 174 };
 175
 176 /* CMM local md_object operations */
 177 static int cml_object_create(const struct lu_context *ctx,
 178                              struct md_object *mo,
 179                              struct lu_attr *attr)
 180 {
 181         int rc;
 182         ENTRY;
 183         rc = mo_object_create(ctx, cmm2child_obj(md2cmm_obj(mo)), attr);
 184         RETURN(rc);
 185 }
 186
 187 static int cml_attr_get(const struct lu_context *ctx, struct md_object *mo,
 188                         struct lu_attr *attr)
 189 {
 190         int rc;
 191         ENTRY;
 192         rc = mo_attr_get(ctx, cmm2child_obj(md2cmm_obj(mo)), attr);
 193         RETURN(rc);
 194 }
 195
 196 static int cml_attr_set(const struct lu_context *ctx, struct md_object *mo,
 197                         struct lu_attr *attr)
 198 {
 199         int rc;
 200         ENTRY;
 201         rc = mo_attr_set(ctx, cmm2child_obj(md2cmm_obj(mo)), attr);
 202         RETURN(rc);
 203 }
 204
 205 static int cml_xattr_get(const struct lu_context *ctx, struct md_object *mo,
 206                          void *buf, int buflen, const char *name)
 207 {
 208         int rc;
 209         ENTRY;
 210         rc = mo_xattr_get(ctx, cmm2child_obj(md2cmm_obj(mo)),
 211                          buf, buflen, name);
 212         RETURN(rc);
 213 }
 214
 215 static int cml_xattr_set(const struct lu_context *ctx, struct md_object *mo,
 216                          void *buf, int buflen, const char *name)
 217 {
 218         int rc;
 219         ENTRY;
 220         rc = mo_xattr_set(ctx, cmm2child_obj(md2cmm_obj(mo)),
 221                           buf, buflen, name);
 222         RETURN(rc);
 223 }
 224
 225 static int cml_ref_add(const struct lu_context *ctx, struct md_object *mo)
 226 {
 227         int rc;
 228         ENTRY;
 229         rc = mo_ref_add(ctx, cmm2child_obj(md2cmm_obj(mo)));
 230         RETURN(rc);
 231 }
 232
 233 static int cml_ref_del(const struct lu_context *ctx, struct md_object *mo)
 234 {
 235         int rc;
 236         ENTRY;
 237         rc = mo_ref_del(ctx, cmm2child_obj(md2cmm_obj(mo)));
 238         RETURN(rc);
 239 }
 240
 241 static int cml_open(const struct lu_context *ctx, struct md_object *mo)
 242 {
 243         int rc;
 244         ENTRY;
 245         rc = mo_open(ctx, cmm2child_obj(md2cmm_obj(mo)));
 246         RETURN(rc);
 247 }
 248
 249 static int cml_close(const struct lu_context *ctx, struct md_object *mo)
 250 {
 251         int rc;
 252         ENTRY;
 253         rc = mo_close(ctx, cmm2child_obj(md2cmm_obj(mo)));
 254         RETURN(rc);
 255 }
 256
 257 static struct md_object_operations cml_mo_ops = {
 258         .moo_attr_get      = cml_attr_get,
 259         .moo_attr_set      = cml_attr_set,
 260         .moo_xattr_get     = cml_xattr_get,
 261         .moo_xattr_set     = cml_xattr_set,
 262         .moo_object_create = cml_object_create,
 263         .moo_ref_add       = cml_ref_add,
 264         .moo_ref_del       = cml_ref_del,
 265         .moo_open          = cml_open,
 266         .moo_close         = cml_close
 267 };
 268
 269 /* md_dir operations */
 270 static int cml_lookup(const struct lu_context *ctx, struct md_object *mo_p,
 271                       const char *name, struct lu_fid *lf)
 272 {
 273         int rc;
 274         ENTRY;
 275         rc = mdo_lookup(ctx, cmm2child_obj(md2cmm_obj(mo_p)), name, lf);
 276         RETURN(rc);
 277
 278 }
 279
 280 static int cml_create(const struct lu_context *ctx,
 281                       struct md_object *mo_p, const char *name,
 282                       struct md_object *mo_c, struct lu_attr *attr)
 283 {
 284         int rc;
 285         ENTRY;
 286         rc = mdo_create(ctx, cmm2child_obj(md2cmm_obj(mo_p)), name,
 287                         cmm2child_obj(md2cmm_obj(mo_c)), attr);
 288         RETURN(rc);
 289 }
 290
 291 static int cml_link(const struct lu_context *ctx, struct md_object *mo_p,
 292                     struct md_object *mo_s, const char *name)
 293 {
 294         int rc;
 295         ENTRY;
 296         rc = mdo_link(ctx, cmm2child_obj(md2cmm_obj(mo_p)),
 297                       cmm2child_obj(md2cmm_obj(mo_s)), name);
 298         RETURN(rc);
 299 }
 300
 301 static int cml_unlink(const struct lu_context *ctx, struct md_object *mo_p,
 302                       struct md_object *mo_c, const char *name)
 303 {
 304         int rc;
 305         ENTRY;
 306         rc = mdo_unlink(ctx, cmm2child_obj(md2cmm_obj(mo_p)),
 307                         cmm2child_obj(md2cmm_obj(mo_c)), name);
 308         RETURN(rc);
 309 }
 310
 311 static int cml_rename(const struct lu_context *ctx, struct md_object *mo_po,
 312                        struct md_object *mo_pn, const struct lu_fid *lf,
 313                        const char *s_name, struct md_object *mo_t,
 314                        const char *t_name)
 315 {
 316         int rc;
 317         ENTRY;
 318
 319         if (mo_t && !cmm_is_local_obj(md2cmm_obj(mo_t))) {
 320                 /* remote object */
 321                 rc = moo_ref_del(ctx, cmm2child_obj(md2cmm_obj(mo_t)));
 322                 if (rc)
 323                         RETURN(rc);
 324                 mo_t = NULL;
 325         }
 326
 327         rc = mdo_rename(ctx, cmm2child_obj(md2cmm_obj(mo_po)),
 328                         cmm2child_obj(md2cmm_obj(mo_pn)), lf, s_name,
 329                         cmm2child_obj(md2cmm_obj(mo_t)), t_name);
 330
 331         RETURN(rc);
 332 }
 333
 334 static int cml_rename_tgt(const struct lu_context *ctx,
 335                           struct md_object *mo_p, struct md_object *mo_t,
 336                           const struct lu_fid *lf, const char *name)
 337 {
 338         int rc;
 339         ENTRY;
 340
 341         rc = mdo_rename_tgt(ctx, cmm2child_obj(md2cmm_obj(mo_po)),
 342                             cmm2child_obj(md2cmm_obj(mo_t)), lf, name);
 343         RETURN(rc);
 344 }
 345
 346 static struct md_dir_operations cmm_dir_ops = {
 347         .mdo_lookup      = cml_lookup,
 348         .mdo_create      = cml_create,
 349         .mdo_link        = cml_link,
 350         .mdo_unlink      = cml_unlink,
 351         .mdo_rename      = cml_rename,
 352         .mdo_rename_tgt  = cml_rename_tgt,
 353 };
 354
 355 /* -------------------------------------------------------------------
 356  * remote CMM object operations. cmr_...
 357  */
 358 static inline struct cmr_object *lu2cmr_obj(struct lu_object *o)
 359 {
 360         return container_of0(o, struct cmr_object, cmm_obj.cmo_obj.mo_lu);
 361 }
 362 static inline struct cmr_object *md2cmr_obj(struct md_object *mo)
 363 {
 364         return container_of0(mo, struct cmr_object, cmm_obj.cmo_obj);
 365 }
 366 static inline struct cmr_object *cmm2cmr_obj(struct cmm_object *co)
 367 {
 368         return container_of0(co, struct cmr_object, cmm_obj);
 369 }
 370
 371 /* get local child device */
 372 static struct lu_device *cmr_child_dev(struct cmm_device *d, __u32 num)
 373 {
 374         struct lu_device *next = NULL;
 375         struct mdc_device *mdc;
 376
 377         spin_lock(&d->cmm_tgt_guard);
 378         list_for_each_entry(mdc, &d->cmm_targets, mc_linkage) {
 379                 if (mdc->mc_num == num) {
 380                         next = mdc2lu_dev(mdc);
 381                         break;
 382                 }
 383         }
 384         spin_unlock(&d->cmm_tgt_guard);
 385         return next;
 386 }
 387
 388 /* lu_object operations */
 389 static void cmr_object_free(const struct lu_context *ctx,
 390                             struct lu_object *lo)
 391 {
 392         struct cmr_object *cro = lu2cmr_obj(lo);
 393         lu_object_fini(lo);
 394         OBD_FREE_PTR(cro);
 395 }
 396
 397 static int cmr_object_init(const struct lu_context *ctx, struct lu_object *lo)
 398 {
 399         struct cmm_device *cd = lu2cmm_dev(lo->lo_dev);
 400         struct lu_device  *c_dev;
 401         struct lu_object  *c_obj;
 402         const struct lu_fid *fid = lu_object_fid(lo);
 403         int rc;
 404
 405         ENTRY;
 406
 407         c_dev = cmr_child_dev(cd, lu2cmr_obj(lo)->cmo_num);
 408         if (c_dev == NULL) {
 409                 rc = -ENOENT;
 410         } else {
 411                 c_obj = c_dev->ld_ops->ldo_object_alloc(ctx,
 412                                                         lo->lo_header, c_dev);
 413                 if (c_obj != NULL) {
 414                         lu_object_add(lo, c_obj);
 415                         rc = 0;
 416                 } else {
 417                         rc = -ENOMEM;
 418                 }
 419         }
 420
 421         RETURN(rc);
 422 }
 423
 424
 425 static int cmr_object_exists(const struct lu_context *ctx,
 426                              struct lu_object *lo)
 427 {
 428         return lu_object_exists(ctx, lu_object_next(lo));
 429 }
 430
 431 static int cmr_object_print(const struct lu_context *ctx,
 432                             struct seq_file *f, const struct lu_object *lo)
 433 {
 434         return seq_printf(f, LUSTRE_CMM0_NAME"-object@%p", lo);
 435 }
 436
 437 static struct lu_object_operations cml_obj_ops = {
 438         .loo_object_init    = cmr_object_init,
 439         .loo_object_free    = cmr_object_free,
 440         .loo_object_print   = cmr_object_print,
 441         .loo_object_exists  = cmr_object_exists
 442 };
 443
 444 /* CMM remote md_object operations. All are invalid */
 445 static int cmr_object_create(const struct lu_context *ctx,
 446                              struct md_object *mo,
 447                              struct lu_attr *attr)
 448 {
 449         RETURN(-EFAULT);
 450 }
 451
 452 static int cmr_attr_get(const struct lu_context *ctx, struct md_object *mo,
 453                         struct lu_attr *attr)
 454 {
 455         RETURN(-EREMOTE);
 456 }
 457
 458 static int cmr_attr_set(const struct lu_context *ctx, struct md_object *mo,
 459                         struct lu_attr *attr)
 460 {
 461         RETURN(-EFAULT);
 462 }
 463
 464 static int cmr_xattr_get(const struct lu_context *ctx, struct md_object *mo,
 465                          void *buf, int buflen, const char *name)
 466 {
 467         RETURN(-EFAULT);
 468 }
 469
 470 static int cmr_xattr_set(const struct lu_context *ctx, struct md_object *mo,
 471                          void *buf, int buflen, const char *name)
 472 {
 473         RETURN(-EFAULT);
 474 }
 475
 476 static int cmr_ref_add(const struct lu_context *ctx, struct md_object *mo)
 477 {
 478         RETURN(-EFAULT);
 479 }
 480
 481 static int cmr_ref_del(const struct lu_context *ctx, struct md_object *mo)
 482 {
 483         RETURN(-EFAULT);
 484 }
 485
 486 static int cmr_open(const struct lu_context *ctx, struct md_object *mo)
 487 {
 488         RETURN(-EREMOTE);
 489 }
 490
 491 static int cmr_close(const struct lu_context *ctx, struct md_object *mo)
 492 {
 493         RETURN(-EFAULT);
 494 }
 495
 496 static struct md_object_operations cml_mo_ops = {
 497         .moo_attr_get      = cmr_attr_get,
 498         .moo_attr_set      = cmr_attr_set,
 499         .moo_xattr_get     = cmr_xattr_get,
 500         .moo_xattr_set     = cmr_xattr_set,
 501         .moo_object_create = cmr_object_create,
 502         .moo_ref_add       = cmr_ref_add,
 503         .moo_ref_del       = cmr_ref_del,
 504         .moo_open          = cmr_open,
 505         .moo_close         = cmr_close
 506 };
 507
 508 /* remote part of md_dir operations */
 509 static int cmr_lookup(const struct lu_context *ctx, struct md_object *mo_p,
 510                       const char *name, struct lu_fid *lf)
 511 {
 512         /*this can happens while rename()
 513          * If new parent is remote dir, lookup will happens here */
 514
 515         RETURN(-EREMOTE);
 516 }
 517
 518 static int cmr_create(const struct lu_context *ctx,
 519                       struct md_object *mo_p, const char *name,
 520                       struct md_object *mo_c, struct lu_attr *attr)
 521 {
 522         int rc;
 523
 524         ENTRY;
 525
 526         //TODO: check the name isn't exist
 527
 528         /* remote object creation and local name insert */
 529         rc = mo_object_create(ctx, cmm2child_obj(md2cmm_obj(mo_c)), attr);
 530         if (rc == 0) {
 531                 rc = mdo_name_insert(ctx, cmm2child_obj(md2cmm_obj(mo_p)),
 532                                      name, lu_object_fid(&mo_c->mo_lu));
 533         }
 534
 535         RETURN(rc);
 536 }
 537
 538 static int cmr_link(const struct lu_context *ctx, struct md_object *mo_p,
 539                     struct md_object *mo_s, const char *name)
 540 {
 541         int rc;
 542         ENTRY;
 543
 544         //TODO: check the name isn't exist
 545
 546         rc = mo_ref_add(ctx, cmm2child_obj(md2cmm_obj(mo_s)));
 547         if (rc == 0) {
 548                 rc = mdo_name_insert(ctx, cmm2child_obj(md2cmm_obj(mo_p)),
 549                                      name, lu_object_fid(&mo_s->mo_lu));
 550         }
 551
 552         RETURN(rc);
 553 }
 554
 555 static int cmr_unlink(const struct lu_context *ctx, struct md_object *mo_p,
 556                       struct md_object *mo_c, const char *name)
 557 {
 558         int rc;
 559         ENTRY;
 560
 561         rc = mo_ref_del(ctx, cmm2child_obj(md2cmm_obj(mo_c)));
 562         if (rc == 0) {
 563                 rc = mdo_name_remove(ctx, cmm2child_obj(md2cmm_obj(mo_p)),
 564                                      name, lu_object_fid(&mo_c->mo_lu));
 565         }
 566
 567         RETURN(rc);
 568 }
 569
 570 static int cmr_rename(const struct lu_context *ctx, struct md_object *mo_po,
 571                        struct md_object *mo_pn, const struct lu_fid *lf,
 572                        const char *s_name, struct md_object *mo_t,
 573                        const char *t_name)
 574 {
 575         int rc;
 576         ENTRY;
 577
 578         /* the mo_pn is remote directory, so we cannot even know if there is
 579          * mo_t or not. Therefore mo_t is NULL here but remote server should do
 580          * lookup and process this further */
 581
 582         LASSERT(mo_t == NULL);
 583         rc = mdo_rename_tgt(ctx, c_pn, NULL/* mo_t */, lf, t_name);
 584         /* only old name is removed localy */
 585         if (rc == 0)
 586                 rc = mdo_name_destroy(ctx, c_po, s_name);
 587
 588         RETURN(rc);
 589 }
 590
 591 static int cmr_rename_tgt(const struct lu_context *ctx,
 592                           struct md_object *mo_p, struct md_object *mo_t,
 593                           const struct lu_fid *lf, const char *name)
 594 {
 595         int rc;
 596         ENTRY;
 597         /* target object is remote one */
 598         rc = mo_ref_del(ctx, cmm2child_obj(md2cmm_obj(mo_t)));
 599         /* continue locally with name handling only */
 600         rc = mdo_rename_tgt(ctx, cmm2child_obj(md2cmm_obj(mo_po)),
 601                             NULL, lf, name);
 602         RETURN(rc);
 603 }
 604
 605 static struct md_dir_operations cmm_dir_ops = {
 606         .mdo_lookup      = cmr_lookup,
 607         .mdo_create      = cmr_create,
 608         .mdo_link        = cmr_link,
 609         .mdo_unlink      = cmr_unlink,
 610         .mdo_rename      = cmr_rename,
 611         .mdo_rename_tgt  = cmr_rename_tgt,
 612 };
 613
 614 #else /* CMM_CODE */
 615 static struct md_object_operations cmm_mo_ops;
 616 static struct md_dir_operations    cmm_dir_ops;
 617 static struct lu_object_operations cmm_obj_ops;
 618
 619 static int cmm_fld_lookup(const struct lu_fid *fid)
 620 {
 621         int rc;
 622         /* temporary hack for proto mkdir */
 623         rc = (unsigned long)fid_seq(fid) / LUSTRE_SEQ_RANGE;
 624         CWARN("Get MDS %d for sequence: "LPU64"\n", rc, fid_seq(fid));
 625         RETURN(rc);
 626 }
 627
 628 /* get child device by mdsnum*/
 629 static struct lu_device *cmm_get_child(struct cmm_device *d, __u32 num)
 630 {
 631         struct lu_device *next = NULL;
 632         ENTRY;
 633         if (likely(num == d->cmm_local_num)) {
 634                 next = &d->cmm_child->md_lu_dev;
 635         } else {
 636                 struct mdc_device *mdc;
 637                 list_for_each_entry(mdc, &d->cmm_targets, mc_linkage) {
 638                         if (mdc->mc_num == num) {
 639                                 next = mdc2lu_dev(mdc);
 640                                 break;
 641                         }
 642                 }
 643         }
 644         RETURN(next);
 645 }
 646
 647 struct lu_object *cmm_object_alloc(const struct lu_context *ctx,
 648                                    const struct lu_object_header *hdr,
 649                                    struct lu_device *ld)
 650 {
 651         struct cmm_object *co;
 652         struct lu_object  *lo;
 653         ENTRY;
 654
 655         OBD_ALLOC_PTR(co);
 656         if (co != NULL) {
 657                 lo = &co->cmo_obj.mo_lu;
 658                 lu_object_init(lo, NULL, ld);
 659                 co->cmo_obj.mo_ops = &cmm_mo_ops;
 660                 co->cmo_obj.mo_dir_ops = &cmm_dir_ops;
 661                 lo->lo_ops = &cmm_obj_ops;
 662         } else
 663                 lo = NULL;
 664
 665         RETURN(lo);
 666 }
 667
 668 static void cmm_object_free(const struct lu_context *ctx, struct lu_object *lo)
 669 {
 670         struct cmm_object *co = lu2cmm_obj(lo);
 671         lu_object_fini(lo);
 672         OBD_FREE_PTR(co);
 673 }
 674
 675 static int cmm_object_init(const struct lu_context *ctx, struct lu_object *lo)
 676 {
 677         struct cmm_device *cd = lu2cmm_dev(lo->lo_dev);
 678         struct lu_device  *c_dev;
 679         struct lu_object  *c_obj;
 680         const struct lu_fid *fid = lu_object_fid(lo);
 681         int mdsnum, rc;
 682
 683         ENTRY;
 684
 685         /* under device can be MDD or MDC */
 686         mdsnum = cmm_fld_lookup(fid);
 687         c_dev = cmm_get_child(cd, mdsnum);
 688         if (c_dev == NULL) {
 689                 rc = -ENOENT;
 690         } else {
 691                 c_obj = c_dev->ld_ops->ldo_object_alloc(ctx,
 692                                                         lo->lo_header, c_dev);
 693                 if (c_obj != NULL) {
 694                         struct cmm_object *co = lu2cmm_obj(lo);
 695
 696                         lu_object_add(lo, c_obj);
 697                         co->cmo_num = mdsnum;
 698                         rc = 0;
 699                 } else {
 700                         rc = -ENOMEM;
 701                 }
 702         }
 703
 704         RETURN(rc);
 705 }
 706
 707 static int cmm_object_exists(const struct lu_context *ctx, struct lu_object *lo)
 708 {
 709         return lu_object_exists(ctx, lu_object_next(lo));
 710 }
 711
 712 static int cmm_object_print(const struct lu_context *ctx,
 713                             struct seq_file *f, const struct lu_object *lo)
 714 {
 715         return seq_printf(f, LUSTRE_CMM0_NAME"-object@%p", lo);
 716 }
 717
 718 static struct lu_object_operations cmm_obj_ops = {
 719         .loo_object_init    = cmm_object_init,
 720         .loo_object_free    = cmm_object_free,
 721         .loo_object_print   = cmm_object_print,
 722         .loo_object_exists  = cmm_object_exists
 723 };
 724
 725 /* md_object operations */
 726 static int cmm_object_create(const struct lu_context *ctx, struct md_object *mo,
 727                              struct lu_attr *attr)
 728 {
 729         struct md_object  *ch = cmm2child_obj(md2cmm_obj(mo));
 730         int rc;
 731
 732         ENTRY;
 733
 734         LASSERT (cmm_is_local_obj(md2cmm_obj(mo)));
 735
 736         rc = mo_object_create(ctx, ch, attr);
 737
 738         RETURN(rc);
 739 }
 740
 741 static int cmm_attr_get(const struct lu_context *ctx, struct md_object *mo,
 742                         struct lu_attr *attr)
 743 {
 744         struct md_object *ch = cmm2child_obj(md2cmm_obj(mo));
 745         int rc;
 746
 747         ENTRY;
 748
 749         LASSERT (cmm_is_local_obj(md2cmm_obj(mo)));
 750
 751         rc = mo_attr_get(ctx, ch, attr);
 752
 753         RETURN(rc);
 754 }
 755
 756 static struct md_object_operations cmm_mo_ops = {
 757         .moo_attr_get      = cmm_attr_get,
 758         .moo_object_create = cmm_object_create,
 759 };
 760
 761 static int cmm_lookup(const struct lu_context *ctx, struct md_object *mo_p,
 762                       const char *name, struct lu_fid *lf)
 763 {
 764         struct md_object *ch_p = cmm2child_obj(md2cmm_obj(mo_p));
 765         int rc;
 766
 767         ENTRY;
 768
 769         LASSERT(cmm_is_local_obj(md2cmm_obj(mo_p)));
 770
 771         rc = mdo_lookup(ctx, ch_p, name, lf);
 772
 773         RETURN(rc);
 774
 775 }
 776
 777 static int cmm_create(const struct lu_context *ctx,
 778                       struct md_object *mo_p, const char *name,
 779                       struct md_object *mo_c, struct lu_attr *attr)
 780 {
 781         struct md_object *ch_c = cmm2child_obj(md2cmm_obj(mo_c));
 782         struct md_object *ch_p = cmm2child_obj(md2cmm_obj(mo_p));
 783         int rc;
 784
 785         ENTRY;
 786
 787         if (cmm_is_local_obj(md2cmm_obj(mo_c))) {
 788                 rc = mdo_create(ctx, ch_p, name, ch_c, attr);
 789         } else {
 790                 const struct lu_fid *lf = lu_object_fid(&mo_c->mo_lu);
 791
 792                 /* remote object creation and local name insert */
 793                 rc = mo_object_create(ctx, ch_c, attr);
 794                 if (rc == 0) {
 795                         rc = mdo_name_insert(ctx, ch_p, name, lf);
 796                 }
 797         }
 798
 799         RETURN(rc);
 800 }
 801
 802 static int cmm_mkdir(const struct lu_context *ctx, struct lu_attr *attr,
 803                      struct md_object *mo_p, const char *name,
 804                      struct md_object *mo_c)
 805 {
 806         struct md_object *ch_c = cmm2child_obj(md2cmm_obj(mo_c));
 807         struct md_object *ch_p = cmm2child_obj(md2cmm_obj(mo_p));
 808         int rc;
 809
 810         ENTRY;
 811
 812         if (cmm_is_local_obj(md2cmm_obj(mo_c))) {
 813                 /* fully local mkdir */
 814                 rc = mdo_mkdir(ctx, attr, ch_p, name, ch_c);
 815         } else {
 816                 const struct lu_fid *lf = lu_object_fid(&mo_c->mo_lu);
 817
 818                 /* remote object creation and local name insert */
 819                 rc = mo_object_create(ctx, ch_c, attr);
 820                 if (rc == 0) {
 821                         rc = mdo_name_insert(ctx, ch_p, name, lf);
 822                 }
 823         }
 824
 825         RETURN(rc);
 826 }
 827
 828 static struct md_dir_operations cmm_dir_ops = {
 829         .mdo_lookup        = cmm_lookup,
 830         .mdo_mkdir         = cmm_mkdir,
 831         .mdo_create        = cmm_create
 832 };
 833 #endif /* CMM_CODE */
 834
 835