lustre/lod/lod_object.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License version 2 for more details.  A copy is
  14  * included in the COPYING file that accompanied this code.
  15
  16  * You should have received a copy of the GNU General Public License
  17  * along with this program; if not, write to the Free Software
  18  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright  2009 Sun Microsystems, Inc. All rights reserved
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2012, Intel, Inc.
  27  */
  28 /*
  29  * lustre/lod/lod_object.c
  30  *
  31  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
  32  */
  33
  34 #ifndef EXPORT_SYMTAB
  35 # define EXPORT_SYMTAB
  36 #endif
  37 #define DEBUG_SUBSYSTEM S_MDS
  38
  39 #include <obd.h>
  40 #include <obd_class.h>
  41 #include <lustre_ver.h>
  42 #include <obd_support.h>
  43 #include <lprocfs_status.h>
  44
  45 #include <lustre_fid.h>
  46 #include <lustre_param.h>
  47 #include <lustre_fid.h>
  48 #include <obd_lov.h>
  49
  50 #include "lod_internal.h"
  51
  52 extern cfs_mem_cache_t *lod_object_kmem;
  53 static const struct dt_body_operations lod_body_lnk_ops;
  54
  55 static int lod_index_lookup(const struct lu_env *env, struct dt_object *dt,
  56                             struct dt_rec *rec, const struct dt_key *key,
  57                             struct lustre_capa *capa)
  58 {
  59         struct dt_object *next = dt_object_child(dt);
  60         return next->do_index_ops->dio_lookup(env, next, rec, key, capa);
  61 }
  62
  63 static int lod_declare_index_insert(const struct lu_env *env,
  64                                     struct dt_object *dt,
  65                                     const struct dt_rec *rec,
  66                                     const struct dt_key *key,
  67                                     struct thandle *handle)
  68 {
  69         return dt_declare_insert(env, dt_object_child(dt), rec, key, handle);
  70 }
  71
  72 static int lod_index_insert(const struct lu_env *env,
  73                             struct dt_object *dt,
  74                             const struct dt_rec *rec,
  75                             const struct dt_key *key,
  76                             struct thandle *th,
  77                             struct lustre_capa *capa,
  78                             int ign)
  79 {
  80         return dt_insert(env, dt_object_child(dt), rec, key, th, capa, ign);
  81 }
  82
  83 static int lod_declare_index_delete(const struct lu_env *env,
  84                                     struct dt_object *dt,
  85                                     const struct dt_key *key,
  86                                     struct thandle *th)
  87 {
  88         return dt_declare_delete(env, dt_object_child(dt), key, th);
  89 }
  90
  91 static int lod_index_delete(const struct lu_env *env,
  92                             struct dt_object *dt,
  93                             const struct dt_key *key,
  94                             struct thandle *th,
  95                             struct lustre_capa *capa)
  96 {
  97         return dt_delete(env, dt_object_child(dt), key, th, capa);
  98 }
  99
 100 static struct dt_it *lod_it_init(const struct lu_env *env,
 101                                  struct dt_object *dt, __u32 attr,
 102                                  struct lustre_capa *capa)
 103 {
 104         struct dt_object   *next = dt_object_child(dt);
 105
 106         return next->do_index_ops->dio_it.init(env, next, attr, capa);
 107 }
 108
 109 static struct dt_index_operations lod_index_ops = {
 110         .dio_lookup         = lod_index_lookup,
 111         .dio_declare_insert = lod_declare_index_insert,
 112         .dio_insert         = lod_index_insert,
 113         .dio_declare_delete = lod_declare_index_delete,
 114         .dio_delete         = lod_index_delete,
 115         .dio_it     = {
 116                 .init       = lod_it_init,
 117         }
 118 };
 119
 120 static void lod_object_read_lock(const struct lu_env *env,
 121                                  struct dt_object *dt, unsigned role)
 122 {
 123         dt_read_lock(env, dt_object_child(dt), role);
 124 }
 125
 126 static void lod_object_write_lock(const struct lu_env *env,
 127                                   struct dt_object *dt, unsigned role)
 128 {
 129         dt_write_lock(env, dt_object_child(dt), role);
 130 }
 131
 132 static void lod_object_read_unlock(const struct lu_env *env,
 133                                    struct dt_object *dt)
 134 {
 135         dt_read_unlock(env, dt_object_child(dt));
 136 }
 137
 138 static void lod_object_write_unlock(const struct lu_env *env,
 139                                     struct dt_object *dt)
 140 {
 141         dt_write_unlock(env, dt_object_child(dt));
 142 }
 143
 144 static int lod_object_write_locked(const struct lu_env *env,
 145                                    struct dt_object *dt)
 146 {
 147         return dt_write_locked(env, dt_object_child(dt));
 148 }
 149
 150 static int lod_attr_get(const struct lu_env *env,
 151                         struct dt_object *dt,
 152                         struct lu_attr *attr,
 153                         struct lustre_capa *capa)
 154 {
 155         return dt_attr_get(env, dt_object_child(dt), attr, capa);
 156 }
 157
 158 static int lod_declare_attr_set(const struct lu_env *env,
 159                                 struct dt_object *dt,
 160                                 const struct lu_attr *attr,
 161                                 struct thandle *handle)
 162 {
 163         struct dt_object  *next = dt_object_child(dt);
 164         int                rc;
 165         ENTRY;
 166
 167         /*
 168          * declare setattr on the local object
 169          */
 170         rc = dt_declare_attr_set(env, next, attr, handle);
 171
 172         RETURN(rc);
 173 }
 174
 175 static int lod_attr_set(const struct lu_env *env,
 176                         struct dt_object *dt,
 177                         const struct lu_attr *attr,
 178                         struct thandle *handle,
 179                         struct lustre_capa *capa)
 180 {
 181         struct dt_object  *next = dt_object_child(dt);
 182         int                rc;
 183         ENTRY;
 184
 185         /*
 186          * apply changes to the local object
 187          */
 188         rc = dt_attr_set(env, next, attr, handle, capa);
 189         if (rc)
 190                 RETURN(rc);
 191
 192         RETURN(rc);
 193 }
 194
 195 static int lod_xattr_get(const struct lu_env *env, struct dt_object *dt,
 196                          struct lu_buf *buf, const char *name,
 197                          struct lustre_capa *capa)
 198 {
 199         return dt_xattr_get(env, dt_object_child(dt), buf, name, capa);
 200 }
 201
 202 /*
 203  * LOV xattr is a storage for striping, and LOD owns this xattr.
 204  * but LOD allows others to control striping to some extent
 205  * - to reset strping
 206  * - to set new defined striping
 207  * - to set new semi-defined striping
 208  *   - number of stripes is defined
 209  *   - number of stripes + osts are defined
 210  *   - ??
 211  */
 212 static int lod_declare_xattr_set(const struct lu_env *env,
 213                                  struct dt_object *dt,
 214                                  const struct lu_buf *buf,
 215                                  const char *name, int fl,
 216                                  struct thandle *th)
 217 {
 218         struct dt_object *next = dt_object_child(dt);
 219         int               rc;
 220         ENTRY;
 221
 222         rc = dt_declare_xattr_set(env, next, buf, name, fl, th);
 223
 224         RETURN(rc);
 225 }
 226
 227 static int lod_xattr_set(const struct lu_env *env,
 228                          struct dt_object *dt, const struct lu_buf *buf,
 229                          const char *name, int fl, struct thandle *th,
 230                          struct lustre_capa *capa)
 231 {
 232         struct dt_object *next = dt_object_child(dt);
 233         int               rc;
 234         ENTRY;
 235
 236         /*
 237          * behave transparantly for all other EAs
 238          */
 239         rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
 240
 241         RETURN(rc);
 242 }
 243
 244 static int lod_declare_xattr_del(const struct lu_env *env,
 245                                  struct dt_object *dt, const char *name,
 246                                  struct thandle *th)
 247 {
 248         return dt_declare_xattr_del(env, dt_object_child(dt), name, th);
 249 }
 250
 251 static int lod_xattr_del(const struct lu_env *env, struct dt_object *dt,
 252                          const char *name, struct thandle *th,
 253                          struct lustre_capa *capa)
 254 {
 255         return dt_xattr_del(env, dt_object_child(dt), name, th, capa);
 256 }
 257
 258 static int lod_xattr_list(const struct lu_env *env,
 259                           struct dt_object *dt, struct lu_buf *buf,
 260                           struct lustre_capa *capa)
 261 {
 262         return dt_xattr_list(env, dt_object_child(dt), buf, capa);
 263 }
 264
 265 int lod_object_set_pool(struct lod_object *o, char *pool)
 266 {
 267         int len;
 268
 269         if (o->ldo_pool) {
 270                 len = strlen(o->ldo_pool);
 271                 OBD_FREE(o->ldo_pool, len + 1);
 272                 o->ldo_pool = NULL;
 273         }
 274         if (pool) {
 275                 len = strlen(pool);
 276                 OBD_ALLOC(o->ldo_pool, len + 1);
 277                 if (o->ldo_pool == NULL)
 278                         return -ENOMEM;
 279                 strcpy(o->ldo_pool, pool);
 280         }
 281         return 0;
 282 }
 283
 284 /**
 285  * used to transfer default striping data to the object being created
 286  */
 287 static void lod_ah_init(const struct lu_env *env,
 288                         struct dt_allocation_hint *ah,
 289                         struct dt_object *parent,
 290                         struct dt_object *child,
 291                         cfs_umode_t child_mode)
 292 {
 293         struct dt_object  *nextc;
 294         struct dt_object  *nextp = NULL;
 295         struct lod_object *lc;
 296         ENTRY;
 297
 298         LASSERT(child);
 299
 300         if (likely(parent))
 301                 nextp = dt_object_child(parent);
 302
 303         nextc = dt_object_child(child);
 304         lc = lod_dt_obj(child);
 305
 306         LASSERT(lc->ldo_stripenr == 0);
 307         LASSERT(lc->ldo_stripe == NULL);
 308
 309         /*
 310          * local object may want some hints
 311          * in case of late striping creation, ->ah_init()
 312          * can be called with local object existing
 313          */
 314         if (!dt_object_exists(nextc))
 315                 nextc->do_ops->do_ah_init(env, ah, nextp, nextc, child_mode);
 316
 317         EXIT;
 318 }
 319
 320 static int lod_declare_object_create(const struct lu_env *env,
 321                                      struct dt_object *dt,
 322                                      struct lu_attr *attr,
 323                                      struct dt_allocation_hint *hint,
 324                                      struct dt_object_format *dof,
 325                                      struct thandle *th)
 326 {
 327         struct dt_object   *next = dt_object_child(dt);
 328         int                 rc;
 329         ENTRY;
 330
 331         LASSERT(dof);
 332         LASSERT(attr);
 333         LASSERT(th);
 334         LASSERT(!dt_object_exists(next));
 335
 336         /*
 337          * first of all, we declare creation of local object
 338          */
 339         rc = dt_declare_create(env, next, attr, hint, dof, th);
 340         if (rc)
 341                 GOTO(out, rc);
 342
 343         if (dof->dof_type == DFT_SYM)
 344                 dt->do_body_ops = &lod_body_lnk_ops;
 345
 346 out:
 347         RETURN(rc);
 348 }
 349
 350 static int lod_object_create(const struct lu_env *env, struct dt_object *dt,
 351                              struct lu_attr *attr,
 352                              struct dt_allocation_hint *hint,
 353                              struct dt_object_format *dof, struct thandle *th)
 354 {
 355         struct dt_object   *next = dt_object_child(dt);
 356         int                 rc;
 357         ENTRY;
 358
 359         /* create local object */
 360         rc = dt_create(env, next, attr, hint, dof, th);
 361
 362         RETURN(rc);
 363 }
 364
 365 static int lod_declare_object_destroy(const struct lu_env *env,
 366                                       struct dt_object *dt,
 367                                       struct thandle *th)
 368 {
 369         struct dt_object   *next = dt_object_child(dt);
 370         int                 rc;
 371         ENTRY;
 372
 373         /*
 374          * we declare destroy for the local object
 375          */
 376         rc = dt_declare_destroy(env, next, th);
 377         if (rc)
 378                 RETURN(rc);
 379
 380         RETURN(rc);
 381 }
 382
 383 static int lod_object_destroy(const struct lu_env *env,
 384                 struct dt_object *dt, struct thandle *th)
 385 {
 386         struct dt_object  *next = dt_object_child(dt);
 387         int                rc;
 388         ENTRY;
 389
 390         /* destroy local object */
 391         rc = dt_destroy(env, next, th);
 392         if (rc)
 393                 RETURN(rc);
 394
 395         RETURN(rc);
 396 }
 397
 398 static int lod_index_try(const struct lu_env *env, struct dt_object *dt,
 399                          const struct dt_index_features *feat)
 400 {
 401         struct dt_object *next = dt_object_child(dt);
 402         int               rc;
 403         ENTRY;
 404
 405         LASSERT(next->do_ops);
 406         LASSERT(next->do_ops->do_index_try);
 407
 408         rc = next->do_ops->do_index_try(env, next, feat);
 409         if (next->do_index_ops && dt->do_index_ops == NULL) {
 410                 dt->do_index_ops = &lod_index_ops;
 411                 /* XXX: iterators don't accept device, so bypass LOD */
 412                 /* will be fixed with DNE */
 413                 if (lod_index_ops.dio_it.fini == NULL) {
 414                         lod_index_ops.dio_it = next->do_index_ops->dio_it;
 415                         lod_index_ops.dio_it.init = lod_it_init;
 416                 }
 417         }
 418
 419         RETURN(rc);
 420 }
 421
 422 static int lod_declare_ref_add(const struct lu_env *env,
 423                                struct dt_object *dt, struct thandle *th)
 424 {
 425         return dt_declare_ref_add(env, dt_object_child(dt), th);
 426 }
 427
 428 static int lod_ref_add(const struct lu_env *env,
 429                        struct dt_object *dt, struct thandle *th)
 430 {
 431         return dt_ref_add(env, dt_object_child(dt), th);
 432 }
 433
 434 static int lod_declare_ref_del(const struct lu_env *env,
 435                                struct dt_object *dt, struct thandle *th)
 436 {
 437         return dt_declare_ref_del(env, dt_object_child(dt), th);
 438 }
 439
 440 static int lod_ref_del(const struct lu_env *env,
 441                        struct dt_object *dt, struct thandle *th)
 442 {
 443         return dt_ref_del(env, dt_object_child(dt), th);
 444 }
 445
 446 static struct obd_capa *lod_capa_get(const struct lu_env *env,
 447                                      struct dt_object *dt,
 448                                      struct lustre_capa *old, __u64 opc)
 449 {
 450         return dt_capa_get(env, dt_object_child(dt), old, opc);
 451 }
 452
 453 static int lod_object_sync(const struct lu_env *env, struct dt_object *dt)
 454 {
 455         return dt_object_sync(env, dt_object_child(dt));
 456 }
 457
 458 struct dt_object_operations lod_obj_ops = {
 459         .do_read_lock           = lod_object_read_lock,
 460         .do_write_lock          = lod_object_write_lock,
 461         .do_read_unlock         = lod_object_read_unlock,
 462         .do_write_unlock        = lod_object_write_unlock,
 463         .do_write_locked        = lod_object_write_locked,
 464         .do_attr_get            = lod_attr_get,
 465         .do_declare_attr_set    = lod_declare_attr_set,
 466         .do_attr_set            = lod_attr_set,
 467         .do_xattr_get           = lod_xattr_get,
 468         .do_declare_xattr_set   = lod_declare_xattr_set,
 469         .do_xattr_set           = lod_xattr_set,
 470         .do_declare_xattr_del   = lod_declare_xattr_del,
 471         .do_xattr_del           = lod_xattr_del,
 472         .do_xattr_list          = lod_xattr_list,
 473         .do_ah_init             = lod_ah_init,
 474         .do_declare_create      = lod_declare_object_create,
 475         .do_create              = lod_object_create,
 476         .do_declare_destroy     = lod_declare_object_destroy,
 477         .do_destroy             = lod_object_destroy,
 478         .do_index_try           = lod_index_try,
 479         .do_declare_ref_add     = lod_declare_ref_add,
 480         .do_ref_add             = lod_ref_add,
 481         .do_declare_ref_del     = lod_declare_ref_del,
 482         .do_ref_del             = lod_ref_del,
 483         .do_capa_get            = lod_capa_get,
 484         .do_object_sync         = lod_object_sync,
 485 };
 486
 487 static ssize_t lod_read(const struct lu_env *env, struct dt_object *dt,
 488                         struct lu_buf *buf, loff_t *pos,
 489                         struct lustre_capa *capa)
 490 {
 491         struct dt_object *next = dt_object_child(dt);
 492         return next->do_body_ops->dbo_read(env, next, buf, pos, capa);
 493 }
 494
 495 static ssize_t lod_declare_write(const struct lu_env *env,
 496                                  struct dt_object *dt,
 497                                  const loff_t size, loff_t pos,
 498                                  struct thandle *th)
 499 {
 500         return dt_declare_record_write(env, dt_object_child(dt),
 501                                        size, pos, th);
 502 }
 503
 504 static ssize_t lod_write(const struct lu_env *env, struct dt_object *dt,
 505                          const struct lu_buf *buf, loff_t *pos,
 506                          struct thandle *th, struct lustre_capa *capa, int iq)
 507 {
 508         struct dt_object *next = dt_object_child(dt);
 509         LASSERT(next);
 510         return next->do_body_ops->dbo_write(env, next, buf, pos, th, capa, iq);
 511 }
 512
 513 static const struct dt_body_operations lod_body_lnk_ops = {
 514         .dbo_read               = lod_read,
 515         .dbo_declare_write      = lod_declare_write,
 516         .dbo_write              = lod_write
 517 };
 518
 519 static int lod_object_init(const struct lu_env *env, struct lu_object *o,
 520                            const struct lu_object_conf *conf)
 521 {
 522         struct lod_device *d = lu2lod_dev(o->lo_dev);
 523         struct lu_object  *below;
 524         struct lu_device  *under;
 525         ENTRY;
 526
 527         /*
 528          * create local object
 529          */
 530         under = &d->lod_child->dd_lu_dev;
 531         below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
 532         if (below == NULL)
 533                 RETURN(-ENOMEM);
 534
 535         lu_object_add(o, below);
 536
 537         RETURN(0);
 538 }
 539
 540 void lod_object_free_striping(const struct lu_env *env, struct lod_object *o)
 541 {
 542 }
 543
 544 /*
 545  * ->start is called once all slices are initialized, including header's
 546  * cache for mode (object type). using the type we can initialize ops
 547  */
 548 static int lod_object_start(const struct lu_env *env, struct lu_object *o)
 549 {
 550         if (S_ISLNK(o->lo_header->loh_attr & S_IFMT))
 551                 lu2lod_obj(o)->ldo_obj.do_body_ops = &lod_body_lnk_ops;
 552         return 0;
 553 }
 554
 555 static void lod_object_free(const struct lu_env *env, struct lu_object *o)
 556 {
 557         struct lod_object *mo = lu2lod_obj(o);
 558
 559         /*
 560          * release all underlying object pinned
 561          */
 562
 563         lod_object_free_striping(env, mo);
 564
 565         lod_object_set_pool(mo, NULL);
 566
 567         lu_object_fini(o);
 568         OBD_SLAB_FREE_PTR(mo, lod_object_kmem);
 569 }
 570
 571 static void lod_object_release(const struct lu_env *env, struct lu_object *o)
 572 {
 573         /* XXX: shouldn't we release everything here in case if object
 574          * creation failed before? */
 575 }
 576
 577 static int lod_object_print(const struct lu_env *env, void *cookie,
 578                             lu_printer_t p, const struct lu_object *l)
 579 {
 580         struct lod_object *o = lu2lod_obj((struct lu_object *) l);
 581
 582         return (*p)(env, cookie, LUSTRE_LOD_NAME"-object@%p", o);
 583 }
 584
 585 struct lu_object_operations lod_lu_obj_ops = {
 586         .loo_object_init        = lod_object_init,
 587         .loo_object_start       = lod_object_start,
 588         .loo_object_free        = lod_object_free,
 589         .loo_object_release     = lod_object_release,
 590         .loo_object_print       = lod_object_print,
 591 };