lustre/obdclass/lu_object.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * Lustre Object.
   5  *
   6  *  Copyright (C) 2006 Cluster File Systems, Inc.
   7  *
   8  *   This file is part of the Lustre file system, http://www.lustre.org
   9  *   Lustre is a trademark of Cluster File Systems, Inc.
  10  *
  11  *   You may have signed or agreed to another license before downloading
  12  *   this software.  If so, you are bound by the terms and conditions
  13  *   of that agreement, and the following does not apply to you.  See the
  14  *   LICENSE file included with this distribution for more information.
  15  *
  16  *   If you did not agree to a different license, then this copy of Lustre
  17  *   is open source software; you can redistribute it and/or modify it
  18  *   under the terms of version 2 of the GNU General Public License as
  19  *   published by the Free Software Foundation.
  20  *
  21  *   In either case, Lustre is distributed in the hope that it will be
  22  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  23  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  24  *   license text for more details.
  25  *
  26  * These are the only exported functions, they provide some generic
  27  * infrastructure for managing object devices
  28  */
  29
  30 #define DEBUG_SUBSYSTEM S_CLASS
  31 #ifndef EXPORT_SYMTAB
  32 # define EXPORT_SYMTAB
  33 #endif
  34
  35 #include <linux/seq_file.h>
  36 #include <linux/module.h>
  37 #include <linux/obd_support.h>
  38 #include <linux/lu_object.h>
  39
  40 #include <libcfs/list.h>
  41
  42 static void lu_object_free(struct lu_context *ctx, struct lu_object *o);
  43
  44 void lu_object_put(struct lu_context *ctxt, struct lu_object *o)
  45 {
  46         struct lu_object_header *top;
  47         struct lu_site          *site;
  48
  49         top = o->lo_header;
  50         site = o->lo_dev->ld_site;
  51         spin_lock(&site->ls_guard);
  52         if (-- top->loh_ref == 0) {
  53                 list_for_each_entry(o, &top->loh_layers, lo_linkage) {
  54                         if (lu_object_ops(o)->ldo_object_release != NULL)
  55                                 lu_object_ops(o)->ldo_object_release(ctxt, o);
  56                 }
  57                 -- site->ls_busy;
  58                 if (lu_object_is_dying(top)) {
  59                         hlist_del_init(&top->loh_hash);
  60                         list_del_init(&top->loh_lru);
  61                 }
  62         }
  63         spin_unlock(&site->ls_guard);
  64         if (lu_object_is_dying(top))
  65                 /*
  66                  * Object was already removed from hash and lru above, can
  67                  * kill it.
  68                  */
  69                 lu_object_free(ctxt, o);
  70 }
  71 EXPORT_SYMBOL(lu_object_put);
  72
  73 struct lu_object *lu_object_alloc(struct lu_context *ctxt,
  74                                   struct lu_site *s, const struct lu_fid *f)
  75 {
  76         struct lu_object *scan;
  77         struct lu_object *top;
  78         int clean;
  79         int result;
  80
  81         top = s->ls_top_dev->ld_ops->ldo_object_alloc(ctxt, s->ls_top_dev);
  82         if (IS_ERR(top))
  83                 RETURN(top);
  84         *lu_object_fid(top) = *f;
  85         do {
  86                 clean = 1;
  87                 list_for_each_entry(scan,
  88                                     &top->lo_header->loh_layers, lo_linkage) {
  89                         if (scan->lo_flags & LU_OBJECT_ALLOCATED)
  90                                 continue;
  91                         clean = 0;
  92                         scan->lo_header = top->lo_header;
  93                         result = lu_object_ops(scan)->ldo_object_init(ctxt,
  94                                                                       scan);
  95                         if (result != 0) {
  96                                 lu_object_free(ctxt, top);
  97                                 RETURN(ERR_PTR(result));
  98                         }
  99                         scan->lo_flags |= LU_OBJECT_ALLOCATED;
 100                 }
 101         } while (!clean);
 102         s->ls_stats.s_created ++;
 103         RETURN(top);
 104 }
 105
 106 static void lu_object_free(struct lu_context *ctx, struct lu_object *o)
 107 {
 108         struct list_head splice;
 109         struct lu_object *scan;
 110
 111         list_for_each_entry_reverse(scan,
 112                                     &o->lo_header->loh_layers, lo_linkage) {
 113                 if (lu_object_ops(scan)->ldo_object_delete != NULL)
 114                         lu_object_ops(scan)->ldo_object_delete(ctx, scan);
 115         }
 116         -- o->lo_dev->ld_site->ls_total;
 117         INIT_LIST_HEAD(&splice);
 118         list_splice_init(&o->lo_header->loh_layers, &splice);
 119         while (!list_empty(&splice)) {
 120                 o = container_of(splice.next, struct lu_object, lo_linkage);
 121                 list_del_init(&o->lo_linkage);
 122                 LASSERT(lu_object_ops(o)->ldo_object_free != NULL);
 123                 lu_object_ops(o)->ldo_object_free(ctx, o);
 124         }
 125 }
 126
 127 void lu_site_purge(struct lu_context *ctx, struct lu_site *s, int nr)
 128 {
 129         struct list_head         dispose;
 130         struct lu_object_header *h;
 131         struct lu_object_header *temp;
 132
 133         INIT_LIST_HEAD(&dispose);
 134         spin_lock(&s->ls_guard);
 135         list_for_each_entry_safe(h, temp, &s->ls_lru, loh_lru) {
 136                 if (nr-- == 0)
 137                         break;
 138                 if (h->loh_ref > 0)
 139                         continue;
 140                 hlist_del_init(&h->loh_hash);
 141                 list_move(&h->loh_lru, &dispose);
 142         }
 143         spin_unlock(&s->ls_guard);
 144         while (!list_empty(&dispose)) {
 145                 h = container_of(dispose.next,
 146                                  struct lu_object_header, loh_lru);
 147                 list_del_init(&h->loh_lru);
 148                 lu_object_free(ctx, lu_object_top(h));
 149                 s->ls_stats.s_lru_purged ++;
 150         }
 151 }
 152 EXPORT_SYMBOL(lu_site_purge);
 153
 154 int lu_object_print(struct lu_context *ctx,
 155                     struct seq_file *f, const struct lu_object *o)
 156 {
 157         static char ruler[] = "........................................";
 158         const struct lu_object *scan;
 159         int nob;
 160         int depth;
 161
 162         nob = 0;
 163         scan = o;
 164         list_for_each_entry_continue(scan, &o->lo_linkage, lo_linkage) {
 165                 depth = scan->lo_depth;
 166                 if (depth <= o->lo_depth && scan != o)
 167                         break;
 168                 LASSERT(lu_object_ops(scan)->ldo_object_print != NULL);
 169                 nob += seq_printf(f, "%*.*s", depth, depth, ruler);
 170                 nob += lu_object_ops(scan)->ldo_object_print(ctx, f, scan);
 171                 nob += seq_printf(f, "\n");
 172         }
 173         return nob;
 174 }
 175 EXPORT_SYMBOL(lu_object_print);
 176
 177 static struct lu_object *htable_lookup(struct lu_site *s,
 178                                        const struct hlist_head *bucket,
 179                                        const struct lu_fid *f)
 180 {
 181         struct lu_object_header *h;
 182         struct hlist_node *scan;
 183
 184         hlist_for_each_entry(h, scan, bucket, loh_hash) {
 185                 s->ls_stats.s_cache_check ++;
 186                 if (lu_fid_eq(&h->loh_fid, f) && !lu_object_is_dying(h)) {
 187                         /* bump reference count... */
 188                         if (h->loh_ref ++ == 0)
 189                                 ++ s->ls_busy;
 190                         /* and move to the head of the LRU */
 191                         list_move_tail(&h->loh_lru, &s->ls_lru);
 192                         s->ls_stats.s_cache_hit ++;
 193                         return lu_object_top(h);
 194                 }
 195         }
 196         s->ls_stats.s_cache_miss ++;
 197         return NULL;
 198 }
 199
 200 static __u32 fid_hash(const struct lu_fid *f)
 201 {
 202         /* all objects with same id and different versions will belong to same
 203          * collisions list. */
 204         return (fid_seq(f) - 1) * LUSTRE_FID_SEQ_WIDTH + fid_oid(f);
 205 }
 206
 207 struct lu_object *lu_object_find(struct lu_context *ctxt, struct lu_site *s,
 208                                  const struct lu_fid *f)
 209 {
 210         struct lu_object  *o;
 211         struct lu_object  *shadow;
 212         struct hlist_head *bucket;
 213
 214         bucket = s->ls_hash + (fid_hash(f) & s->ls_hash_mask);
 215         spin_lock(&s->ls_guard);
 216         o = htable_lookup(s, bucket, f);
 217         spin_unlock(&s->ls_guard);
 218         if (o != NULL)
 219                 return o;
 220
 221         o = lu_object_alloc(ctxt, s, f);
 222         if (IS_ERR(o))
 223                 return o;
 224
 225         ++ s->ls_total;
 226         LASSERT(lu_fid_eq(lu_object_fid(o), f));
 227
 228         spin_lock(&s->ls_guard);
 229         shadow = htable_lookup(s, bucket, f);
 230         if (shadow == NULL) {
 231                 hlist_add_head(&o->lo_header->loh_hash, bucket);
 232                 list_add_tail(&s->ls_lru, &o->lo_header->loh_lru);
 233                 shadow = o;
 234                 o = NULL;
 235         } else
 236                 s->ls_stats.s_cache_race ++;
 237         spin_unlock(&s->ls_guard);
 238         if (o != NULL)
 239                 lu_object_free(ctxt, o);
 240         return shadow;
 241 }
 242 EXPORT_SYMBOL(lu_object_find);
 243
 244 enum {
 245         LU_SITE_HTABLE_BITS = 8,
 246         LU_SITE_HTABLE_SIZE = (1 << LU_SITE_HTABLE_BITS),
 247         LU_SITE_HTABLE_MASK = LU_SITE_HTABLE_SIZE - 1
 248 };
 249
 250 int lu_site_init(struct lu_site *s, struct lu_device *top)
 251 {
 252         memset(s, 0, sizeof *s);
 253
 254         spin_lock_init(&s->ls_guard);
 255         CFS_INIT_LIST_HEAD(&s->ls_lru);
 256         s->ls_top_dev = top;
 257         top->ld_site = s;
 258         lu_device_get(top);
 259         /*
 260          * XXX nikita: fixed size hash-table.
 261          */
 262         s->ls_hash_mask = LU_SITE_HTABLE_MASK;
 263         OBD_ALLOC(s->ls_hash, LU_SITE_HTABLE_SIZE * sizeof s->ls_hash[0]);
 264         if (s->ls_hash != NULL) {
 265                 int i;
 266                 for (i = 0; i < LU_SITE_HTABLE_SIZE; i++)
 267                         INIT_HLIST_HEAD(&s->ls_hash[i]);
 268                 return 0;
 269         } else
 270                 return -ENOMEM;
 271 }
 272 EXPORT_SYMBOL(lu_site_init);
 273
 274 void lu_site_fini(struct lu_site *s)
 275 {
 276         LASSERT(list_empty(&s->ls_lru));
 277         LASSERT(s->ls_total == 0);
 278         LASSERT(s->ls_busy == 0);
 279
 280         if (s->ls_hash != NULL) {
 281                 int i;
 282                 for (i = 0; i < LU_SITE_HTABLE_SIZE; i++)
 283                         LASSERT(hlist_empty(&s->ls_hash[i]));
 284                 OBD_FREE(s->ls_hash,
 285                          LU_SITE_HTABLE_SIZE * sizeof s->ls_hash[0]);
 286                 s->ls_hash = NULL;
 287        }
 288        if (s->ls_top_dev != NULL) {
 289                lu_device_put(s->ls_top_dev);
 290                s->ls_top_dev->ld_site = NULL;
 291                s->ls_top_dev = NULL;
 292        }
 293  }
 294 EXPORT_SYMBOL(lu_site_fini);
 295
 296 void lu_device_get(struct lu_device *d)
 297 {
 298         atomic_inc(&d->ld_ref);
 299 }
 300 EXPORT_SYMBOL(lu_device_get);
 301
 302 void lu_device_put(struct lu_device *d)
 303 {
 304         atomic_dec(&d->ld_ref);
 305 }
 306 EXPORT_SYMBOL(lu_device_put);
 307
 308 int lu_device_init(struct lu_device *d, struct lu_device_type *t)
 309 {
 310         memset(d, 0, sizeof *d);
 311         atomic_set(&d->ld_ref, 0);
 312         d->ld_type = t;
 313         return 0;
 314 }
 315 EXPORT_SYMBOL(lu_device_init);
 316
 317 void lu_device_fini(struct lu_device *d)
 318 {
 319         LASSERT(atomic_read(&d->ld_ref) == 0);
 320 }
 321 EXPORT_SYMBOL(lu_device_fini);
 322
 323 int lu_object_init(struct lu_object *o,
 324                    struct lu_object_header *h, struct lu_device *d)
 325 {
 326         memset(o, 0, sizeof *o);
 327         o->lo_header = h;
 328         o->lo_dev    = d;
 329         lu_device_get(d);
 330         CFS_INIT_LIST_HEAD(&o->lo_linkage);
 331         return 0;
 332 }
 333 EXPORT_SYMBOL(lu_object_init);
 334
 335 void lu_object_fini(struct lu_object *o)
 336 {
 337         LASSERT(list_empty(&o->lo_linkage));
 338
 339         if (o->lo_dev != NULL) {
 340                 lu_device_get(o->lo_dev);
 341                 o->lo_dev = NULL;
 342         }
 343 }
 344 EXPORT_SYMBOL(lu_object_fini);
 345
 346 void lu_object_add_top(struct lu_object_header *h, struct lu_object *o)
 347 {
 348         list_move(&o->lo_linkage, &h->loh_layers);
 349 }
 350 EXPORT_SYMBOL(lu_object_add_top);
 351
 352 void lu_object_add(struct lu_object *before, struct lu_object *o)
 353 {
 354         list_move(&o->lo_linkage, &before->lo_linkage);
 355 }
 356 EXPORT_SYMBOL(lu_object_add);
 357
 358 int lu_object_header_init(struct lu_object_header *h)
 359 {
 360         memset(h, 0, sizeof *h);
 361         INIT_HLIST_NODE(&h->loh_hash);
 362         CFS_INIT_LIST_HEAD(&h->loh_lru);
 363         CFS_INIT_LIST_HEAD(&h->loh_layers);
 364         return 0;
 365 }
 366 EXPORT_SYMBOL(lu_object_header_init);
 367
 368 void lu_object_header_fini(struct lu_object_header *h)
 369 {
 370         LASSERT(list_empty(&h->loh_layers));
 371         LASSERT(list_empty(&h->loh_lru));
 372         LASSERT(hlist_unhashed(&h->loh_hash));
 373 }
 374 EXPORT_SYMBOL(lu_object_header_fini);
 375
 376 struct lu_object *lu_object_locate(struct lu_object_header *h,
 377                                    struct lu_device_type *dtype)
 378 {
 379         struct lu_object *o;
 380
 381         list_for_each_entry(o, &h->loh_layers, lo_linkage) {
 382                 if (o->lo_dev->ld_type == dtype)
 383                         return o;
 384         }
 385         return NULL;
 386 }
 387 EXPORT_SYMBOL(lu_object_locate);
 388
 389 enum {
 390         /*
 391          * Maximal number of tld slots.
 392          */
 393         LU_CONTEXT_KEY_NR = 16
 394 };
 395
 396 static struct lu_context_key *lu_keys[LU_CONTEXT_KEY_NR] = { NULL, };
 397
 398 static spinlock_t lu_keys_guard = SPIN_LOCK_UNLOCKED;
 399
 400 int lu_context_key_register(struct lu_context_key *key)
 401 {
 402         int result;
 403         int i;
 404
 405         result = -ENFILE;
 406         spin_lock(&lu_keys_guard);
 407         for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
 408                 if (lu_keys[i] == NULL) {
 409                         key->lct_index = i;
 410                         key->lct_used = 1;
 411                         lu_keys[i] = key;
 412                         result = 0;
 413                         break;
 414                 }
 415         }
 416         spin_unlock(&lu_keys_guard);
 417         return result;
 418 }
 419 EXPORT_SYMBOL(lu_context_key_register);
 420
 421 void lu_context_key_degister(struct lu_context_key *key)
 422 {
 423         LASSERT(key->lct_used >= 1);
 424         LASSERT(0 <= key->lct_index && key->lct_index < ARRAY_SIZE(lu_keys));
 425
 426         if (key->lct_used > 1)
 427                 CERROR("key has instances.\n");
 428         spin_lock(&lu_keys_guard);
 429         lu_keys[key->lct_index] = NULL;
 430         spin_unlock(&lu_keys_guard);
 431 }
 432 EXPORT_SYMBOL(lu_context_key_degister);
 433
 434 void *lu_context_key_get(struct lu_context *ctx, struct lu_context_key *key)
 435 {
 436         LASSERT(0 <= key->lct_index && key->lct_index < ARRAY_SIZE(lu_keys));
 437         return ctx->lc_value[key->lct_index];
 438 }
 439 EXPORT_SYMBOL(lu_context_key_get);
 440
 441 static void keys_fini(struct lu_context *ctx)
 442 {
 443         int i;
 444
 445         if (ctx->lc_value != NULL) {
 446                 for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
 447                         if (ctx->lc_value[i] != NULL) {
 448                                 struct lu_context_key *key;
 449
 450                                 key = lu_keys[i];
 451                                 LASSERT(key != NULL);
 452                                 LASSERT(key->lct_fini != NULL);
 453                                 LASSERT(key->lct_used > 1);
 454
 455                                 key->lct_fini(ctx, ctx->lc_value[i]);
 456                                 key->lct_used--;
 457                                 ctx->lc_value[i] = NULL;
 458                         }
 459                 }
 460                 OBD_FREE(ctx->lc_value,
 461                          ARRAY_SIZE(lu_keys) * sizeof ctx->lc_value[0]);
 462                 ctx->lc_value = NULL;
 463         }
 464 }
 465
 466 static int keys_init(struct lu_context *ctx)
 467 {
 468         int i;
 469         int result;
 470
 471         OBD_ALLOC(ctx->lc_value, ARRAY_SIZE(lu_keys) * sizeof ctx->lc_value[0]);
 472         if (ctx->lc_value != NULL) {
 473                 for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
 474                         struct lu_context_key *key;
 475
 476                         key = lu_keys[i];
 477                         if (key != NULL) {
 478                                 void *value;
 479
 480                                 LASSERT(key->lct_init != NULL);
 481                                 LASSERT(key->lct_index == i);
 482
 483                                 value = key->lct_init(ctx);
 484                                 if (IS_ERR(value)) {
 485                                         keys_fini(ctx);
 486                                         return PTR_ERR(value);
 487                                 }
 488                                 key->lct_used++;
 489                                 ctx->lc_value[i] = value;
 490                         }
 491                 }
 492                 result = 0;
 493         } else
 494                 result = -ENOMEM;
 495         return result;
 496 }
 497
 498 int lu_context_init(struct lu_context *ctx)
 499 {
 500         memset(ctx, 0, sizeof *ctx);
 501         keys_init(ctx);
 502         return 0;
 503 }
 504 EXPORT_SYMBOL(lu_context_init);
 505
 506 void lu_context_fini(struct lu_context *ctx)
 507 {
 508         keys_fini(ctx);
 509 }
 510 EXPORT_SYMBOL(lu_context_fini);
 511
 512 void lu_context_enter(struct lu_context *ctx)
 513 {
 514 }
 515 EXPORT_SYMBOL(lu_context_enter);
 516
 517 void lu_context_exit(struct lu_context *ctx)
 518 {
 519 }
 520 EXPORT_SYMBOL(lu_context_exit);