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, struct md_object *mo_s,
 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)),
 329                         cmm2child_obj(md2cmm_obj(mo_s)), s_name,
 330                         cmm2child_obj(md2cmm_obj(mo_t)), t_name);
 331         RETURN(rc);
 332 }
 333
 334 static int cml_rename_tgt(const struct lu_context *ctx,
 335                           struct md_object *mo_p,
 336                           struct md_object *mo_s, struct md_object *mo_t,
 337                           const char *name)
 338 {
 339         int rc;
 340         ENTRY;
 341
 342         rc = mdo_rename_tgt(ctx, cmm2child_obj(md2cmm_obj(mo_po)),
 343                             cmm2child_obj(md2cmm_obj(mo_s)),
 344                             cmm2child_obj(md2cmm_obj(mo_t)), name);
 345         RETURN(rc);
 346 }
 347
 348 static int cml_name_insert(const struct lu_context * ctx,
 349                            struct md_object *mo_p,
 350                            const char *name, const struct lu_fid *lf)
 351 {
 352         int rc;
 353         ENTRY;
 354
 355         rc = mdo_name_insert(ctx, cmm2child_obj(md2cmm_obj(mo_po)), name, lf);
 356
 357         RETURN(rc);
 358 }
 359
 360 static struct md_dir_operations cmm_dir_ops = {
 361         .mdo_lookup      = cml_lookup,
 362         .mdo_create      = cml_create,
 363         .mdo_link        = cml_link,
 364         .mdo_unlink      = cml_unlink,
 365         .mdo_rename      = cml_rename,
 366         .mdo_rename_tgt  = cml_rename_tgt,
 367         .mdo_name_insert = cml_name_insert
 368 };
 369
 370 /* -------------------------------------------------------------------
 371  * remote CMM object operations. cmr_...
 372  */
 373 static inline struct cmr_object *lu2cmr_obj(struct lu_object *o)
 374 {
 375         return container_of0(o, struct cmr_object, cmm_obj.cmo_obj.mo_lu);
 376 }
 377 static inline struct cmr_object *md2cmr_obj(struct md_object *mo)
 378 {
 379         return container_of0(mo, struct cmr_object, cmm_obj.cmo_obj);
 380 }
 381 static inline struct cmr_object *cmm2cmr_obj(struct cmm_object *co)
 382 {
 383         return container_of0(co, struct cmr_object, cmm_obj);
 384 }
 385
 386 /* get local child device */
 387 static struct lu_device *cmr_child_dev(struct cmm_device *d, __u32 num)
 388 {
 389         struct lu_device *next = NULL;
 390         struct mdc_device *mdc;
 391
 392         spin_lock(&d->cmm_tgt_guard);
 393         list_for_each_entry(mdc, &d->cmm_targets, mc_linkage) {
 394                 if (mdc->mc_num == num) {
 395                         next = mdc2lu_dev(mdc);
 396                         break;
 397                 }
 398         }
 399         spin_unlock(&d->cmm_tgt_guard);
 400         return next;
 401 }
 402
 403 /* lu_object operations */
 404 static void cmr_object_free(const struct lu_context *ctx,
 405                             struct lu_object *lo)
 406 {
 407         struct cmr_object *cro = lu2cmr_obj(lo);
 408         lu_object_fini(lo);
 409         OBD_FREE_PTR(cro);
 410 }
 411
 412 static int cmr_object_init(const struct lu_context *ctx, struct lu_object *lo)
 413 {
 414         struct cmm_device *cd = lu2cmm_dev(lo->lo_dev);
 415         struct lu_device  *c_dev;
 416         struct lu_object  *c_obj;
 417         const struct lu_fid *fid = lu_object_fid(lo);
 418         int rc;
 419
 420         ENTRY;
 421
 422         c_dev = cmr_child_dev(cd, lu2cmr_obj(lo)->cmo_num);
 423         if (c_dev == NULL) {
 424                 rc = -ENOENT;
 425         } else {
 426                 c_obj = c_dev->ld_ops->ldo_object_alloc(ctx,
 427                                                         lo->lo_header, c_dev);
 428                 if (c_obj != NULL) {
 429                         lu_object_add(lo, c_obj);
 430                         rc = 0;
 431                 } else {
 432                         rc = -ENOMEM;
 433                 }
 434         }
 435
 436         RETURN(rc);
 437 }
 438
 439
 440 static int cmr_object_exists(const struct lu_context *ctx,
 441                              struct lu_object *lo)
 442 {
 443         return lu_object_exists(ctx, lu_object_next(lo));
 444 }
 445
 446 static int cmr_object_print(const struct lu_context *ctx,
 447                             struct seq_file *f, const struct lu_object *lo)
 448 {
 449         return seq_printf(f, LUSTRE_CMM0_NAME"-object@%p", lo);
 450 }
 451
 452 static struct lu_object_operations cml_obj_ops = {
 453         .loo_object_init    = cmr_object_init,
 454         .loo_object_free    = cmr_object_free,
 455         .loo_object_print   = cmr_object_print,
 456         .loo_object_exists  = cmr_object_exists
 457 };
 458
 459 /* CMM remote md_object operations. All are invalid */
 460 static int cmr_object_create(const struct lu_context *ctx,
 461                              struct md_object *mo,
 462                              struct lu_attr *attr)
 463 {
 464         RETURN(-EFAULT);
 465 }
 466
 467 static int cmr_attr_get(const struct lu_context *ctx, struct md_object *mo,
 468                         struct lu_attr *attr)
 469 {
 470         RETURN(-EREMOTE);
 471 }
 472
 473 static int cmr_attr_set(const struct lu_context *ctx, struct md_object *mo,
 474                         struct lu_attr *attr)
 475 {
 476         RETURN(-EFAULT);
 477 }
 478
 479 static int cmr_xattr_get(const struct lu_context *ctx, struct md_object *mo,
 480                          void *buf, int buflen, const char *name)
 481 {
 482         RETURN(-EFAULT);
 483 }
 484
 485 static int cmr_xattr_set(const struct lu_context *ctx, struct md_object *mo,
 486                          void *buf, int buflen, const char *name)
 487 {
 488         RETURN(-EFAULT);
 489 }
 490
 491 static int cmr_ref_add(const struct lu_context *ctx, struct md_object *mo)
 492 {
 493         RETURN(-EFAULT);
 494 }
 495
 496 static int cmr_ref_del(const struct lu_context *ctx, struct md_object *mo)
 497 {
 498         RETURN(-EFAULT);
 499 }
 500
 501 static int cmr_open(const struct lu_context *ctx, struct md_object *mo)
 502 {
 503         RETURN(-EREMOTE);
 504 }
 505
 506 static int cmr_close(const struct lu_context *ctx, struct md_object *mo)
 507 {
 508         RETURN(-EFAULT);
 509 }
 510
 511 static struct md_object_operations cml_mo_ops = {
 512         .moo_attr_get      = cmr_attr_get,
 513         .moo_attr_set      = cmr_attr_set,
 514         .moo_xattr_get     = cmr_xattr_get,
 515         .moo_xattr_set     = cmr_xattr_set,
 516         .moo_object_create = cmr_object_create,
 517         .moo_ref_add       = cmr_ref_add,
 518         .moo_ref_del       = cmr_ref_del,
 519         .moo_open          = cmr_open,
 520         .moo_close         = cmr_close
 521 };
 522
 523 /* remote part of md_dir operations */
 524 static int cmr_lookup(const struct lu_context *ctx, struct md_object *mo_p,
 525                       const char *name, struct lu_fid *lf)
 526 {
 527         RETURN(-EFAULT);
 528 }
 529
 530 static int cmr_create(const struct lu_context *ctx,
 531                       struct md_object *mo_p, const char *name,
 532                       struct md_object *mo_c, struct lu_attr *attr)
 533 {
 534         int rc;
 535
 536         ENTRY;
 537
 538         //TODO: check the name isn't exist
 539
 540         /* remote object creation and local name insert */
 541         rc = mo_object_create(ctx, cmm2child_obj(md2cmm_obj(mo_c)), attr);
 542         if (rc == 0) {
 543                 rc = mdo_name_insert(ctx, cmm2child_obj(md2cmm_obj(mo_p)),
 544                                      name, lu_object_fid(&mo_c->mo_lu));
 545         }
 546
 547         RETURN(rc);
 548 }
 549
 550 static int cmr_link(const struct lu_context *ctx, struct md_object *mo_p,
 551                     struct md_object *mo_s, const char *name)
 552 {
 553         int rc;
 554         ENTRY;
 555
 556         //TODO: check the name isn't exist
 557
 558         rc = mo_ref_add(ctx, cmm2child_obj(md2cmm_obj(mo_s)));
 559         if (rc == 0) {
 560                 rc = mdo_name_insert(ctx, cmm2child_obj(md2cmm_obj(mo_p)),
 561                                      name, lu_object_fid(&mo_s->mo_lu));
 562         }
 563
 564         RETURN(rc);
 565 }
 566
 567 static int cmr_unlink(const struct lu_context *ctx, struct md_object *mo_p,
 568                       struct md_object *mo_c, const char *name)
 569 {
 570         int rc;
 571         ENTRY;
 572
 573         rc = mo_ref_del(ctx, cmm2child_obj(md2cmm_obj(mo_c)));
 574         if (rc == 0) {
 575                 rc = mdo_name_remove(ctx, cmm2child_obj(md2cmm_obj(mo_p)),
 576                                      name, lu_object_fid(&mo_c->mo_lu));
 577         }
 578
 579         RETURN(rc);
 580 }
 581
 582 static int cmr_rename(const struct lu_context *ctx, struct md_object *mo_po,
 583                        struct md_object *mo_pn, struct md_object *mo_s,
 584                        const char *s_name, struct md_object *mo_t,
 585                        const char *t_name)
 586 {
 587         int rc;
 588         ENTRY;
 589         rc = mdo_rename(ctx, cmm2child_obj(md2cmm_obj(mo_po)),
 590                         cmm2child_obj(md2cmm_obj(mo_pn)),
 591                         cmm2child_obj(md2cmm_obj(mo_s)), s_name,
 592                         cmm2child_obj(md2cmm_obj(mo_t)), t_name);
 593
 594         if (mo_t == NULL) {
 595                 rc = mdo_name_insert(ctx, c_pn, t_name,
 596                                      lu_object_fid(&c_s->mo_lu));
 597                 rc = mdo_name_remove(ctx, c_po, s_name);
 598         } else {
 599                 c_t =  cmm2child_obj(md2cmm_obj(mo_t));
 600                 if (cmm_is_local_obj(md2cmm_obj(mo_t))) {
 601                         /*
 602                          * target object is local so only name should
 603                          * deleted/inserted on remote server
 604                          */
 605                         rc = mdo_rename_tgt(ctx, c_pn, c_s,
 606                                             NULL, t_name);
 607                         /* localy the old name will be removed and target object
 608                          * will be destroeyd*/
 609                         rc = mdo_rename(ctx, c_po, NULL, c_s,
 610                                         s_name, c_t, NULL);
 611                } else {
 612                         /* target object is remote one so just ask remote server
 613                          * to continue with rename */
 614                         rc = mdo_rename_tgt(ctx, c_pn, c_s,
 615                                             c_t, t_name);
 616                         /* only old name is removed localy */
 617                         rc = mdo_name_destroy(ctx, c_po, s_name);
 618                }
 619        }
 620
 621         RETURN(rc);
 622 }
 623
 624 static int cmr_rename_tgt(const struct lu_context *ctx,
 625                           struct md_object *mo_p,
 626                           struct md_object *mo_s, struct md_object *mo_t,
 627                           const char *name)
 628 {
 629         int rc;
 630         ENTRY;
 631         /* target object is remote one */
 632         rc = mo_ref_del(ctx, cmm2child_obj(md2cmm_obj(mo_t)));
 633         /* continue locally with name handling only */
 634         rc = mdo_rename_tgt(ctx, cmm2child_obj(md2cmm_obj(mo_po)),
 635                             cmm2child_obj(md2cmm_obj(mo_s)),
 636                             NULL, name);
 637         RETURN(rc);
 638 }
 639
 640 static int cmr_name_insert(const struct lu_context * ctx,
 641                            struct md_object *mo_p,
 642                            const char *name, const struct lu_fid *lf)
 643 {
 644         RETURN(-EFAULT);
 645 }
 646
 647 static struct md_dir_operations cmm_dir_ops = {
 648         .mdo_lookup      = cmr_lookup,
 649         .mdo_create      = cmr_create,
 650         .mdo_link        = cmr_link,
 651         .mdo_unlink      = cmr_unlink,
 652         .mdo_rename      = cmr_rename,
 653         .mdo_rename_tgt  = cmr_rename_tgt,
 654         .mdo_name_insert = cmr_name_insert,
 655 };
 656
 657 #else /* CMM_CODE */
 658 static struct md_object_operations cmm_mo_ops;
 659 static struct md_dir_operations    cmm_dir_ops;
 660 static struct lu_object_operations cmm_obj_ops;
 661
 662 static int cmm_fld_lookup(const struct lu_fid *fid)
 663 {
 664         int rc;
 665         /* temporary hack for proto mkdir */
 666         rc = (unsigned long)fid_seq(fid) / LUSTRE_SEQ_RANGE;
 667         CWARN("Get MDS %d for sequence: "LPU64"\n", rc, fid_seq(fid));
 668         RETURN(rc);
 669 }
 670
 671 /* get child device by mdsnum*/
 672 static struct lu_device *cmm_get_child(struct cmm_device *d, __u32 num)
 673 {
 674         struct lu_device *next = NULL;
 675         ENTRY;
 676         if (likely(num == d->cmm_local_num)) {
 677                 next = &d->cmm_child->md_lu_dev;
 678         } else {
 679                 struct mdc_device *mdc;
 680                 list_for_each_entry(mdc, &d->cmm_targets, mc_linkage) {
 681                         if (mdc->mc_num == num) {
 682                                 next = mdc2lu_dev(mdc);
 683                                 break;
 684                         }
 685                 }
 686         }
 687         RETURN(next);
 688 }
 689
 690 struct lu_object *cmm_object_alloc(const struct lu_context *ctx,
 691                                    const struct lu_object_header *hdr,
 692                                    struct lu_device *ld)
 693 {
 694         struct cmm_object *co;
 695         struct lu_object  *lo;
 696         ENTRY;
 697
 698         OBD_ALLOC_PTR(co);
 699         if (co != NULL) {
 700                 lo = &co->cmo_obj.mo_lu;
 701                 lu_object_init(lo, NULL, ld);
 702                 co->cmo_obj.mo_ops = &cmm_mo_ops;
 703                 co->cmo_obj.mo_dir_ops = &cmm_dir_ops;
 704                 lo->lo_ops = &cmm_obj_ops;
 705         } else
 706                 lo = NULL;
 707
 708         RETURN(lo);
 709 }
 710
 711 static void cmm_object_free(const struct lu_context *ctx, struct lu_object *lo)
 712 {
 713         struct cmm_object *co = lu2cmm_obj(lo);
 714         lu_object_fini(lo);
 715         OBD_FREE_PTR(co);
 716 }
 717
 718 static int cmm_object_init(const struct lu_context *ctx, struct lu_object *lo)
 719 {
 720         struct cmm_device *cd = lu2cmm_dev(lo->lo_dev);
 721         struct lu_device  *c_dev;
 722         struct lu_object  *c_obj;
 723         const struct lu_fid *fid = lu_object_fid(lo);
 724         int mdsnum, rc;
 725
 726         ENTRY;
 727
 728         /* under device can be MDD or MDC */
 729         mdsnum = cmm_fld_lookup(fid);
 730         c_dev = cmm_get_child(cd, mdsnum);
 731         if (c_dev == NULL) {
 732                 rc = -ENOENT;
 733         } else {
 734                 c_obj = c_dev->ld_ops->ldo_object_alloc(ctx,
 735                                                         lo->lo_header, c_dev);
 736                 if (c_obj != NULL) {
 737                         struct cmm_object *co = lu2cmm_obj(lo);
 738
 739                         lu_object_add(lo, c_obj);
 740                         co->cmo_num = mdsnum;
 741                         rc = 0;
 742                 } else {
 743                         rc = -ENOMEM;
 744                 }
 745         }
 746
 747         RETURN(rc);
 748 }
 749
 750 static int cmm_object_exists(const struct lu_context *ctx, struct lu_object *lo)
 751 {
 752         return lu_object_exists(ctx, lu_object_next(lo));
 753 }
 754
 755 static int cmm_object_print(const struct lu_context *ctx,
 756                             struct seq_file *f, const struct lu_object *lo)
 757 {
 758         return seq_printf(f, LUSTRE_CMM0_NAME"-object@%p", lo);
 759 }
 760
 761 static struct lu_object_operations cmm_obj_ops = {
 762         .loo_object_init    = cmm_object_init,
 763         .loo_object_free    = cmm_object_free,
 764         .loo_object_print   = cmm_object_print,
 765         .loo_object_exists  = cmm_object_exists
 766 };
 767
 768 /* md_object operations */
 769 static int cmm_object_create(const struct lu_context *ctx, struct md_object *mo,
 770                              struct lu_attr *attr)
 771 {
 772         struct md_object  *ch = cmm2child_obj(md2cmm_obj(mo));
 773         int rc;
 774
 775         ENTRY;
 776
 777         LASSERT (cmm_is_local_obj(md2cmm_obj(mo)));
 778
 779         rc = mo_object_create(ctx, ch, attr);
 780
 781         RETURN(rc);
 782 }
 783
 784 static int cmm_attr_get(const struct lu_context *ctx, struct md_object *mo,
 785                         struct lu_attr *attr)
 786 {
 787         struct md_object *ch = cmm2child_obj(md2cmm_obj(mo));
 788         int rc;
 789
 790         ENTRY;
 791
 792         LASSERT (cmm_is_local_obj(md2cmm_obj(mo)));
 793
 794         rc = mo_attr_get(ctx, ch, attr);
 795
 796         RETURN(rc);
 797 }
 798
 799 static struct md_object_operations cmm_mo_ops = {
 800         .moo_attr_get      = cmm_attr_get,
 801         .moo_object_create = cmm_object_create,
 802 };
 803
 804 static int cmm_lookup(const struct lu_context *ctx, struct md_object *mo_p,
 805                       const char *name, struct lu_fid *lf)
 806 {
 807         struct md_object *ch_p = cmm2child_obj(md2cmm_obj(mo_p));
 808         int rc;
 809
 810         ENTRY;
 811
 812         LASSERT(cmm_is_local_obj(md2cmm_obj(mo_p)));
 813
 814         rc = mdo_lookup(ctx, ch_p, name, lf);
 815
 816         RETURN(rc);
 817
 818 }
 819
 820 static int cmm_create(const struct lu_context *ctx,
 821                       struct md_object *mo_p, const char *name,
 822                       struct md_object *mo_c, struct lu_attr *attr)
 823 {
 824         struct md_object *ch_c = cmm2child_obj(md2cmm_obj(mo_c));
 825         struct md_object *ch_p = cmm2child_obj(md2cmm_obj(mo_p));
 826         int rc;
 827
 828         ENTRY;
 829
 830         if (cmm_is_local_obj(md2cmm_obj(mo_c))) {
 831                 rc = mdo_create(ctx, ch_p, name, ch_c, attr);
 832         } else {
 833                 const struct lu_fid *lf = lu_object_fid(&mo_c->mo_lu);
 834
 835                 /* remote object creation and local name insert */
 836                 rc = mo_object_create(ctx, ch_c, attr);
 837                 if (rc == 0) {
 838                         rc = mdo_name_insert(ctx, ch_p, name, lf);
 839                 }
 840         }
 841
 842         RETURN(rc);
 843 }
 844
 845 static int cmm_mkdir(const struct lu_context *ctx, struct lu_attr *attr,
 846                      struct md_object *mo_p, const char *name,
 847                      struct md_object *mo_c)
 848 {
 849         struct md_object *ch_c = cmm2child_obj(md2cmm_obj(mo_c));
 850         struct md_object *ch_p = cmm2child_obj(md2cmm_obj(mo_p));
 851         int rc;
 852
 853         ENTRY;
 854
 855         if (cmm_is_local_obj(md2cmm_obj(mo_c))) {
 856                 /* fully local mkdir */
 857                 rc = mdo_mkdir(ctx, attr, ch_p, name, ch_c);
 858         } else {
 859                 const struct lu_fid *lf = lu_object_fid(&mo_c->mo_lu);
 860
 861                 /* remote object creation and local name insert */
 862                 rc = mo_object_create(ctx, ch_c, attr);
 863                 if (rc == 0) {
 864                         rc = mdo_name_insert(ctx, ch_p, name, lf);
 865                 }
 866         }
 867
 868         RETURN(rc);
 869 }
 870
 871 static struct md_dir_operations cmm_dir_ops = {
 872         .mdo_lookup        = cmm_lookup,
 873         .mdo_mkdir         = cmm_mkdir,
 874         .mdo_create        = cmm_create
 875 };
 876 #endif /* CMM_CODE */
 877
 878