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 {
  42         int rc = 0;
  43         ENTRY;
  44
  45         LASSERT(fid_is_sane(fid));
  46
  47         rc = fld_client_lookup(&cm->cmm_fld, fid_seq(fid), mds);
  48         if (rc) {
  49                 CERROR("can't find mds by seq "LPU64", rc %d\n",
  50                        fid_seq(fid), rc);
  51                 RETURN(rc);
  52         }
  53
  54         if (*mds > cm->cmm_tgt_count) {
  55                 CERROR("Got invalid mdsno: "LPU64" (max: %u)\n",
  56                        *mds, cm->cmm_tgt_count);
  57                 rc = -EINVAL;
  58         } else {
  59                 CDEBUG(D_INFO, "CMM: got MDS "LPU64" for sequence: "LPU64"\n",
  60                        *mds, fid_seq(fid));
  61         }
  62
  63         RETURN (rc);
  64 }
  65
  66 static struct md_object_operations cml_mo_ops;
  67 static struct md_dir_operations    cml_dir_ops;
  68 static struct lu_object_operations cml_obj_ops;
  69
  70 static struct md_object_operations cmr_mo_ops;
  71 static struct md_dir_operations    cmr_dir_ops;
  72 static struct lu_object_operations cmr_obj_ops;
  73
  74 struct lu_object *cmm_object_alloc(const struct lu_context *ctx,
  75                                    const struct lu_object_header *loh,
  76                                    struct lu_device *ld)
  77 {
  78         struct lu_object  *lo = NULL;
  79         const struct lu_fid *fid = &loh->loh_fid;
  80         struct cmm_device *cd;
  81         mdsno_t mdsnum;
  82         int rc = 0;
  83
  84         ENTRY;
  85
  86         cd = lu2cmm_dev(ld);
  87         if (cd->cmm_flags & CMM_INITIALIZED) {
  88                 /* get object location */
  89                 rc = cmm_fld_lookup(lu2cmm_dev(ld), fid, &mdsnum);
  90                 if (rc)
  91                         RETURN(ERR_PTR(rc));
  92         } else
  93                 /*
  94                  * Device is not yet initialized, cmm_object is being created
  95                  * as part of early bootstrap procedure (it is /ROOT, or /fld,
  96                  * etc.). Such object *has* to be local.
  97                  */
  98                 mdsnum = cd->cmm_local_num;
  99
 100         /* select the proper set of operations based on object location */
 101         if (mdsnum == cd->cmm_local_num) {
 102                 struct cml_object *clo;
 103
 104                 OBD_ALLOC_PTR(clo);
 105                 if (clo != NULL) {
 106                         lo = &clo->cmm_obj.cmo_obj.mo_lu;
 107                         lu_object_init(lo, NULL, ld);
 108                         clo->cmm_obj.cmo_obj.mo_ops = &cml_mo_ops;
 109                         clo->cmm_obj.cmo_obj.mo_dir_ops = &cml_dir_ops;
 110                         lo->lo_ops = &cml_obj_ops;
 111                         clo->cmm_obj.cmo_local = 1;
 112                 }
 113         } else {
 114                 struct cmr_object *cro;
 115
 116                 OBD_ALLOC_PTR(cro);
 117                 if (cro != NULL) {
 118                         lo = &cro->cmm_obj.cmo_obj.mo_lu;
 119                         lu_object_init(lo, NULL, ld);
 120                         cro->cmm_obj.cmo_obj.mo_ops = &cmr_mo_ops;
 121                         cro->cmm_obj.cmo_obj.mo_dir_ops = &cmr_dir_ops;
 122                         lo->lo_ops = &cmr_obj_ops;
 123                         cro->cmo_num = mdsnum;
 124                         cro->cmm_obj.cmo_local = 0;
 125                 }
 126         }
 127         RETURN(lo);
 128 }
 129
 130 /*
 131  * CMM has two types of objects - local and remote. They have different set
 132  * of operations so we are avoiding multiple checks in code.
 133  */
 134
 135 /*
 136  * local CMM object operations. cml_...
 137  */
 138 static inline struct cml_object *lu2cml_obj(struct lu_object *o)
 139 {
 140         return container_of0(o, struct cml_object, cmm_obj.cmo_obj.mo_lu);
 141 }
 142 static inline struct cml_object *md2cml_obj(struct md_object *mo)
 143 {
 144         return container_of0(mo, struct cml_object, cmm_obj.cmo_obj);
 145 }
 146 static inline struct cml_object *cmm2cml_obj(struct cmm_object *co)
 147 {
 148         return container_of0(co, struct cml_object, cmm_obj);
 149 }
 150 /* get local child device */
 151 static struct lu_device *cml_child_dev(struct cmm_device *d)
 152 {
 153         return &d->cmm_child->md_lu_dev;
 154 }
 155
 156 /* lu_object operations */
 157 static void cml_object_free(const struct lu_context *ctx,
 158                             struct lu_object *lo)
 159 {
 160         struct cml_object *clo = lu2cml_obj(lo);
 161         lu_object_fini(lo);
 162         OBD_FREE_PTR(clo);
 163 }
 164
 165 static int cml_object_init(const struct lu_context *ctx, struct lu_object *lo)
 166 {
 167         struct cmm_device *cd = lu2cmm_dev(lo->lo_dev);
 168         struct lu_device  *c_dev;
 169         struct lu_object  *c_obj;
 170         int rc;
 171
 172         ENTRY;
 173
 174         c_dev = cml_child_dev(cd);
 175         if (c_dev == NULL) {
 176                 rc = -ENOENT;
 177         } else {
 178                 c_obj = c_dev->ld_ops->ldo_object_alloc(ctx,
 179                                                         lo->lo_header, c_dev);
 180                 if (c_obj != NULL) {
 181                         lu_object_add(lo, c_obj);
 182                         rc = 0;
 183                 } else {
 184                         rc = -ENOMEM;
 185                 }
 186         }
 187
 188         RETURN(rc);
 189 }
 190
 191 static int cml_object_exists(const struct lu_context *ctx,
 192                              struct lu_object *lo)
 193 {
 194         return lu_object_exists(ctx, lu_object_next(lo));
 195 }
 196
 197 static int cml_object_print(const struct lu_context *ctx,
 198                             struct seq_file *f, const struct lu_object *lo)
 199 {
 200         return seq_printf(f, LUSTRE_CMM0_NAME"-local@%p", lo);
 201 }
 202
 203 static struct lu_object_operations cml_obj_ops = {
 204         .loo_object_init    = cml_object_init,
 205         .loo_object_free    = cml_object_free,
 206         .loo_object_print   = cml_object_print,
 207         .loo_object_exists  = cml_object_exists
 208 };
 209
 210 /* CMM local md_object operations */
 211 static int cml_object_create(const struct lu_context *ctx,
 212                              struct md_object *mo,
 213                              struct lu_attr *attr)
 214 {
 215         int rc;
 216         ENTRY;
 217         rc = mo_object_create(ctx, md_object_next(mo), attr);
 218         RETURN(rc);
 219 }
 220
 221 static int cml_attr_get(const struct lu_context *ctx, struct md_object *mo,
 222                         struct lu_attr *attr)
 223 {
 224         int rc;
 225         ENTRY;
 226         rc = mo_attr_get(ctx, md_object_next(mo), attr);
 227         RETURN(rc);
 228 }
 229
 230 static int cml_attr_set(const struct lu_context *ctx, struct md_object *mo,
 231                         const struct lu_attr *attr)
 232 {
 233         int rc;
 234         ENTRY;
 235         rc = mo_attr_set(ctx, md_object_next(mo), attr);
 236         RETURN(rc);
 237 }
 238
 239 static int cml_xattr_get(const struct lu_context *ctx, struct md_object *mo,
 240                          void *buf, int buflen, const char *name)
 241 {
 242         int rc;
 243         ENTRY;
 244         rc = mo_xattr_get(ctx, md_object_next(mo),
 245                          buf, buflen, name);
 246         RETURN(rc);
 247 }
 248
 249 static int cml_xattr_set(const struct lu_context *ctx, struct md_object *mo,
 250                          const void *buf, int buflen, const char *name)
 251 {
 252         int rc;
 253         ENTRY;
 254         rc = mo_xattr_set(ctx, md_object_next(mo),
 255                           buf, buflen, name);
 256         RETURN(rc);
 257 }
 258
 259 static int cml_ref_add(const struct lu_context *ctx, struct md_object *mo)
 260 {
 261         int rc;
 262         ENTRY;
 263         rc = mo_ref_add(ctx, md_object_next(mo));
 264         RETURN(rc);
 265 }
 266
 267 static int cml_ref_del(const struct lu_context *ctx, struct md_object *mo)
 268 {
 269         int rc;
 270         ENTRY;
 271         rc = mo_ref_del(ctx, md_object_next(mo));
 272         RETURN(rc);
 273 }
 274
 275 static int cml_open(const struct lu_context *ctx, struct md_object *mo)
 276 {
 277         int rc;
 278         ENTRY;
 279         rc = mo_open(ctx, md_object_next(mo));
 280         RETURN(rc);
 281 }
 282
 283 static int cml_close(const struct lu_context *ctx, struct md_object *mo)
 284 {
 285         int rc;
 286         ENTRY;
 287         rc = mo_close(ctx, md_object_next(mo));
 288         RETURN(rc);
 289 }
 290
 291 static struct md_object_operations cml_mo_ops = {
 292         .moo_attr_get      = cml_attr_get,
 293         .moo_attr_set      = cml_attr_set,
 294         .moo_xattr_get     = cml_xattr_get,
 295         .moo_xattr_set     = cml_xattr_set,
 296         .moo_object_create = cml_object_create,
 297         .moo_ref_add       = cml_ref_add,
 298         .moo_ref_del       = cml_ref_del,
 299         .moo_open          = cml_open,
 300         .moo_close         = cml_close
 301 };
 302
 303 /* md_dir operations */
 304 static int cml_lookup(const struct lu_context *ctx, struct md_object *mo_p,
 305                       const char *name, struct lu_fid *lf)
 306 {
 307         int rc;
 308         ENTRY;
 309         rc = mdo_lookup(ctx, md_object_next(mo_p), name, lf);
 310         RETURN(rc);
 311
 312 }
 313
 314 static int cml_create(const struct lu_context *ctx,
 315                       struct md_object *mo_p, const char *name,
 316                       struct md_object *mo_c, struct lu_attr *attr)
 317 {
 318         int rc;
 319         ENTRY;
 320         rc = mdo_create(ctx, md_object_next(mo_p), name,
 321                         md_object_next(mo_c), attr);
 322         RETURN(rc);
 323 }
 324
 325 static int cml_link(const struct lu_context *ctx, struct md_object *mo_p,
 326                     struct md_object *mo_s, const char *name)
 327 {
 328         int rc;
 329         ENTRY;
 330         rc = mdo_link(ctx, md_object_next(mo_p),
 331                       md_object_next(mo_s), name);
 332         RETURN(rc);
 333 }
 334
 335 static int cml_unlink(const struct lu_context *ctx, struct md_object *mo_p,
 336                       struct md_object *mo_c, const char *name)
 337 {
 338         int rc;
 339         ENTRY;
 340         rc = mdo_unlink(ctx, md_object_next(mo_p),
 341                         md_object_next(mo_c), name);
 342         RETURN(rc);
 343 }
 344
 345 /* rename is split to local/remote by location of new parent dir */
 346 static int cml_rename(const struct lu_context *ctx, struct md_object *mo_po,
 347                        struct md_object *mo_pn, const struct lu_fid *lf,
 348                        const char *s_name, struct md_object *mo_t,
 349                        const char *t_name)
 350 {
 351         int rc;
 352         ENTRY;
 353
 354         if (mo_t && !cmm_is_local_obj(md2cmm_obj(mo_t))) {
 355                 /* mo_t is remote object and there is RPC to unlink it */
 356                 rc = mo_ref_del(ctx, md_object_next(mo_t));
 357                 if (rc)
 358                         RETURN(rc);
 359                 mo_t = NULL;
 360         }
 361         /* local rename, mo_t can be NULL */
 362         rc = mdo_rename(ctx, md_object_next(mo_po),
 363                         md_object_next(mo_pn), lf, s_name,
 364                         md_object_next(mo_t), t_name);
 365
 366         RETURN(rc);
 367 }
 368
 369 static int cml_rename_tgt(const struct lu_context *ctx,
 370                           struct md_object *mo_p, struct md_object *mo_t,
 371                           const struct lu_fid *lf, const char *name)
 372 {
 373         int rc;
 374         ENTRY;
 375
 376         rc = mdo_rename_tgt(ctx, md_object_next(mo_p),
 377                             md_object_next(mo_t), lf, name);
 378         RETURN(rc);
 379 }
 380
 381 static struct md_dir_operations cml_dir_ops = {
 382         .mdo_lookup      = cml_lookup,
 383         .mdo_create      = cml_create,
 384         .mdo_link        = cml_link,
 385         .mdo_unlink      = cml_unlink,
 386         .mdo_rename      = cml_rename,
 387         .mdo_rename_tgt  = cml_rename_tgt,
 388 };
 389
 390 /* -------------------------------------------------------------------
 391  * remote CMM object operations. cmr_...
 392  */
 393 static inline struct cmr_object *lu2cmr_obj(struct lu_object *o)
 394 {
 395         return container_of0(o, struct cmr_object, cmm_obj.cmo_obj.mo_lu);
 396 }
 397 static inline struct cmr_object *md2cmr_obj(struct md_object *mo)
 398 {
 399         return container_of0(mo, struct cmr_object, cmm_obj.cmo_obj);
 400 }
 401 static inline struct cmr_object *cmm2cmr_obj(struct cmm_object *co)
 402 {
 403         return container_of0(co, struct cmr_object, cmm_obj);
 404 }
 405
 406 /* get proper child device from MDCs */
 407 static struct lu_device *cmr_child_dev(struct cmm_device *d, __u32 num)
 408 {
 409         struct lu_device *next = NULL;
 410         struct mdc_device *mdc;
 411
 412         spin_lock(&d->cmm_tgt_guard);
 413         list_for_each_entry(mdc, &d->cmm_targets, mc_linkage) {
 414                 if (mdc->mc_num == num) {
 415                         next = mdc2lu_dev(mdc);
 416                         break;
 417                 }
 418         }
 419         spin_unlock(&d->cmm_tgt_guard);
 420         return next;
 421 }
 422
 423 /* lu_object operations */
 424 static void cmr_object_free(const struct lu_context *ctx,
 425                             struct lu_object *lo)
 426 {
 427         struct cmr_object *cro = lu2cmr_obj(lo);
 428         lu_object_fini(lo);
 429         OBD_FREE_PTR(cro);
 430 }
 431
 432 static int cmr_object_init(const struct lu_context *ctx, struct lu_object *lo)
 433 {
 434         struct cmm_device *cd = lu2cmm_dev(lo->lo_dev);
 435         struct lu_device  *c_dev;
 436         struct lu_object  *c_obj;
 437         int rc;
 438
 439         ENTRY;
 440
 441         c_dev = cmr_child_dev(cd, lu2cmr_obj(lo)->cmo_num);
 442         if (c_dev == NULL) {
 443                 rc = -ENOENT;
 444         } else {
 445                 c_obj = c_dev->ld_ops->ldo_object_alloc(ctx,
 446                                                         lo->lo_header, c_dev);
 447                 if (c_obj != NULL) {
 448                         lu_object_add(lo, c_obj);
 449                         rc = 0;
 450                 } else {
 451                         rc = -ENOMEM;
 452                 }
 453         }
 454
 455         RETURN(rc);
 456 }
 457
 458 static int cmr_object_exists(const struct lu_context *ctx,
 459                              struct lu_object *lo)
 460 {
 461         return lu_object_exists(ctx, lu_object_next(lo));
 462 }
 463
 464 static int cmr_object_print(const struct lu_context *ctx,
 465                             struct seq_file *f, const struct lu_object *lo)
 466 {
 467         return seq_printf(f, LUSTRE_CMM0_NAME"-remote@%p", lo);
 468 }
 469
 470 static struct lu_object_operations cmr_obj_ops = {
 471         .loo_object_init    = cmr_object_init,
 472         .loo_object_free    = cmr_object_free,
 473         .loo_object_print   = cmr_object_print,
 474         .loo_object_exists  = cmr_object_exists
 475 };
 476
 477 /* CMM remote md_object operations. All are invalid */
 478 static int cmr_object_create(const struct lu_context *ctx,
 479                              struct md_object *mo,
 480                              struct lu_attr *attr)
 481 {
 482         RETURN(-EFAULT);
 483 }
 484
 485 static int cmr_attr_get(const struct lu_context *ctx, struct md_object *mo,
 486                         struct lu_attr *attr)
 487 {
 488         RETURN(-EREMOTE);
 489 }
 490
 491 static int cmr_attr_set(const struct lu_context *ctx, struct md_object *mo,
 492                         const struct lu_attr *attr)
 493 {
 494         RETURN(-EFAULT);
 495 }
 496
 497 static int cmr_xattr_get(const struct lu_context *ctx, struct md_object *mo,
 498                          void *buf, int buflen, const char *name)
 499 {
 500         RETURN(-EFAULT);
 501 }
 502
 503 static int cmr_xattr_set(const struct lu_context *ctx, struct md_object *mo,
 504                          const void *buf, int buflen, const char *name)
 505 {
 506         RETURN(-EFAULT);
 507 }
 508
 509 static int cmr_ref_add(const struct lu_context *ctx, struct md_object *mo)
 510 {
 511         RETURN(-EFAULT);
 512 }
 513
 514 static int cmr_ref_del(const struct lu_context *ctx, struct md_object *mo)
 515 {
 516         RETURN(-EFAULT);
 517 }
 518
 519 static int cmr_open(const struct lu_context *ctx, struct md_object *mo)
 520 {
 521         RETURN(-EREMOTE);
 522 }
 523
 524 static int cmr_close(const struct lu_context *ctx, struct md_object *mo)
 525 {
 526         RETURN(-EFAULT);
 527 }
 528
 529 static struct md_object_operations cmr_mo_ops = {
 530         .moo_attr_get      = cmr_attr_get,
 531         .moo_attr_set      = cmr_attr_set,
 532         .moo_xattr_get     = cmr_xattr_get,
 533         .moo_xattr_set     = cmr_xattr_set,
 534         .moo_object_create = cmr_object_create,
 535         .moo_ref_add       = cmr_ref_add,
 536         .moo_ref_del       = cmr_ref_del,
 537         .moo_open          = cmr_open,
 538         .moo_close         = cmr_close
 539 };
 540
 541 /* remote part of md_dir operations */
 542 static int cmr_lookup(const struct lu_context *ctx, struct md_object *mo_p,
 543                       const char *name, struct lu_fid *lf)
 544 {
 545         /*this can happens while rename()
 546          * If new parent is remote dir, lookup will happens here */
 547
 548         RETURN(-EREMOTE);
 549 }
 550
 551 /*
 552  * All methods below are cross-ref by nature. They consist of remote call and
 553  * local operation. Due to future rollback functionality there are several
 554  * limitations for such methods:
 555  * 1) remote call should be done at first to do epoch negotiation between all
 556  * MDS involved and to avoid the RPC inside transaction.
 557  * 2) only one RPC can be sent - also due to epoch negotiation.
 558  * For more details see rollback HLD/DLD.
 559  *
 560  */
 561 static int cmr_create(const struct lu_context *ctx,
 562                       struct md_object *mo_p, const char *name,
 563                       struct md_object *mo_c, struct lu_attr *attr)
 564 {
 565         int rc;
 566
 567         ENTRY;
 568
 569         //XXX: make sure that MDT checks name isn't exist
 570
 571         /* remote object creation and local name insert */
 572         rc = mo_object_create(ctx, md_object_next(mo_c), attr);
 573         if (rc == 0) {
 574                 rc = mdo_name_insert(ctx, md_object_next(mo_p),
 575                                      name, lu_object_fid(&mo_c->mo_lu));
 576         }
 577
 578         RETURN(rc);
 579 }
 580
 581 static int cmr_link(const struct lu_context *ctx, struct md_object *mo_p,
 582                     struct md_object *mo_s, const char *name)
 583 {
 584         int rc;
 585         ENTRY;
 586
 587         //XXX: make sure that MDT checks name isn't exist
 588
 589         rc = mo_ref_add(ctx, md_object_next(mo_s));
 590         if (rc == 0) {
 591                 rc = mdo_name_insert(ctx, md_object_next(mo_p),
 592                                      name, lu_object_fid(&mo_s->mo_lu));
 593         }
 594
 595         RETURN(rc);
 596 }
 597
 598 static int cmr_unlink(const struct lu_context *ctx, struct md_object *mo_p,
 599                       struct md_object *mo_c, const char *name)
 600 {
 601         int rc;
 602         ENTRY;
 603
 604         rc = mo_ref_del(ctx, md_object_next(mo_c));
 605         if (rc == 0) {
 606                 rc = mdo_name_remove(ctx, md_object_next(mo_p),
 607                                      name);
 608         }
 609
 610         RETURN(rc);
 611 }
 612
 613 static int cmr_rename(const struct lu_context *ctx, struct md_object *mo_po,
 614                        struct md_object *mo_pn, const struct lu_fid *lf,
 615                        const char *s_name, struct md_object *mo_t,
 616                        const char *t_name)
 617 {
 618         int rc;
 619         ENTRY;
 620
 621         /* the mo_pn is remote directory, so we cannot even know if there is
 622          * mo_t or not. Therefore mo_t is NULL here but remote server should do
 623          * lookup and process this further */
 624
 625         LASSERT(mo_t == NULL);
 626         rc = mdo_rename_tgt(ctx, md_object_next(mo_pn),
 627                             NULL/* mo_t */, lf, t_name);
 628         /* only old name is removed localy */
 629         if (rc == 0)
 630                 rc = mdo_name_remove(ctx, md_object_next(mo_po),
 631                                      s_name);
 632
 633         RETURN(rc);
 634 }
 635
 636 /* part of cross-ref rename(). Used to insert new name in new parent
 637  * and unlink target with same name if it exists */
 638 static int cmr_rename_tgt(const struct lu_context *ctx,
 639                           struct md_object *mo_p, struct md_object *mo_t,
 640                           const struct lu_fid *lf, const char *name)
 641 {
 642         int rc;
 643         ENTRY;
 644         /* target object is remote one */
 645         rc = mo_ref_del(ctx, md_object_next(mo_t));
 646         /* continue locally with name handling only */
 647         if (rc == 0)
 648                 rc = mdo_rename_tgt(ctx, md_object_next(mo_p),
 649                                     NULL, lf, name);
 650         RETURN(rc);
 651 }
 652
 653 static struct md_dir_operations cmr_dir_ops = {
 654         .mdo_lookup      = cmr_lookup,
 655         .mdo_create      = cmr_create,
 656         .mdo_link        = cmr_link,
 657         .mdo_unlink      = cmr_unlink,
 658         .mdo_rename      = cmr_rename,
 659         .mdo_rename_tgt  = cmr_rename_tgt,
 660 };
 661
 662