lustre/lmv/lmv_object.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  */
  36
  37 #ifndef EXPORT_SYMTAB
  38 # define EXPORT_SYMTAB
  39 #endif
  40 #define DEBUG_SUBSYSTEM S_LMV
  41 #ifdef __KERNEL__
  42 #include <linux/slab.h>
  43 #include <linux/module.h>
  44 #include <linux/init.h>
  45 #include <linux/slab.h>
  46 #include <linux/pagemap.h>
  47 #include <asm/div64.h>
  48 #include <linux/seq_file.h>
  49 #else
  50 #include <liblustre.h>
  51 #endif
  52
  53 #include <lustre/lustre_idl.h>
  54 #include <obd_support.h>
  55 #include <lustre_lib.h>
  56 #include <lustre_net.h>
  57 #include <lustre_dlm.h>
  58 #include <obd_class.h>
  59 #include <lprocfs_status.h>
  60 #include "lmv_internal.h"
  61
  62 extern cfs_mem_cache_t *lmv_object_cache;
  63 extern atomic_t lmv_object_count;
  64
  65 static CFS_LIST_HEAD(obj_list);
  66 static spinlock_t obj_list_lock = SPIN_LOCK_UNLOCKED;
  67
  68 struct lmv_object *lmv_object_alloc(struct obd_device *obd,
  69                                     const struct lu_fid *fid,
  70                                     struct lmv_stripe_md *mea)
  71 {
  72         struct lmv_obd          *lmv = &obd->u.lmv;
  73         unsigned int             obj_size;
  74         struct lmv_object       *obj;
  75         int                      i;
  76
  77         LASSERT(mea->mea_magic == MEA_MAGIC_LAST_CHAR
  78                 || mea->mea_magic == MEA_MAGIC_ALL_CHARS
  79                 || mea->mea_magic == MEA_MAGIC_HASH_SEGMENT);
  80
  81         OBD_SLAB_ALLOC(obj, lmv_object_cache, CFS_ALLOC_STD,
  82                        sizeof(*obj));
  83         if (!obj)
  84                 return NULL;
  85
  86         atomic_inc(&lmv_object_count);
  87
  88         obj->lo_fid = *fid;
  89         obj->lo_obd = obd;
  90         obj->lo_state = 0;
  91         obj->lo_hashtype = mea->mea_magic;
  92
  93         init_MUTEX(&obj->lo_guard);
  94         atomic_set(&obj->lo_count, 0);
  95         obj->lo_objcount = mea->mea_count;
  96
  97         obj_size = sizeof(struct lmv_stripe) *
  98                 lmv->desc.ld_tgt_count;
  99
 100         OBD_ALLOC(obj->lo_stripes, obj_size);
 101         if (!obj->lo_stripes)
 102                 goto err_obj;
 103
 104         memset(obj->lo_stripes, 0, obj_size);
 105
 106         CDEBUG(D_INODE, "Allocate object for "DFID"\n",
 107                PFID(fid));
 108         for (i = 0; i < mea->mea_count; i++) {
 109                 int rc;
 110
 111                 CDEBUG(D_INODE, "Process subobject "DFID"\n",
 112                        PFID(&mea->mea_ids[i]));
 113                 obj->lo_stripes[i].ls_fid = mea->mea_ids[i];
 114                 LASSERT(fid_is_sane(&obj->lo_stripes[i].ls_fid));
 115
 116                 /*
 117                  * Cache slave mds number to use it in all cases it is needed
 118                  * instead of constant lookup.
 119                  */
 120                 rc = lmv_fld_lookup(lmv, &obj->lo_stripes[i].ls_fid,
 121                                     &obj->lo_stripes[i].ls_mds);
 122                 if (rc)
 123                         goto err_obj;
 124         }
 125
 126         return obj;
 127 err_obj:
 128         OBD_FREE(obj, sizeof(*obj));
 129         return NULL;
 130 }
 131
 132 void lmv_object_free(struct lmv_object *obj)
 133 {
 134         struct lmv_obd          *lmv = &obj->lo_obd->u.lmv;
 135         unsigned int             obj_size;
 136
 137         LASSERT(!atomic_read(&obj->lo_count));
 138
 139         obj_size = sizeof(struct lmv_stripe) *
 140                 lmv->desc.ld_tgt_count;
 141
 142         OBD_FREE(obj->lo_stripes, obj_size);
 143         OBD_SLAB_FREE(obj, lmv_object_cache, sizeof(*obj));
 144         atomic_dec(&lmv_object_count);
 145 }
 146
 147 static void __lmv_object_add(struct lmv_object *obj)
 148 {
 149         atomic_inc(&obj->lo_count);
 150         list_add(&obj->lo_list, &obj_list);
 151 }
 152
 153 void lmv_object_add(struct lmv_object *obj)
 154 {
 155         spin_lock(&obj_list_lock);
 156         __lmv_object_add(obj);
 157         spin_unlock(&obj_list_lock);
 158 }
 159
 160 static void __lmv_object_del(struct lmv_object *obj)
 161 {
 162         list_del(&obj->lo_list);
 163         lmv_object_free(obj);
 164 }
 165
 166 void lmv_object_del(struct lmv_object *obj)
 167 {
 168         spin_lock(&obj_list_lock);
 169         __lmv_object_del(obj);
 170         spin_unlock(&obj_list_lock);
 171 }
 172
 173 static struct lmv_object *__lmv_object_get(struct lmv_object *obj)
 174 {
 175         LASSERT(obj != NULL);
 176         atomic_inc(&obj->lo_count);
 177         return obj;
 178 }
 179
 180 struct lmv_object *lmv_object_get(struct lmv_object *obj)
 181 {
 182         spin_lock(&obj_list_lock);
 183         __lmv_object_get(obj);
 184         spin_unlock(&obj_list_lock);
 185         return obj;
 186 }
 187
 188 static void __lmv_object_put(struct lmv_object *obj)
 189 {
 190         LASSERT(obj);
 191
 192         if (atomic_dec_and_test(&obj->lo_count)) {
 193                 CDEBUG(D_INODE, "Last reference to "DFID" - "
 194                        "destroying\n", PFID(&obj->lo_fid));
 195                 __lmv_object_del(obj);
 196         }
 197 }
 198
 199 void lmv_object_put(struct lmv_object *obj)
 200 {
 201         spin_lock(&obj_list_lock);
 202         __lmv_object_put(obj);
 203         spin_unlock(&obj_list_lock);
 204 }
 205
 206 void lmv_object_put_unlock(struct lmv_object *obj)
 207 {
 208         lmv_object_unlock(obj);
 209         lmv_object_put(obj);
 210 }
 211
 212 static struct lmv_object *__lmv_object_find(struct obd_device *obd, const struct lu_fid *fid)
 213 {
 214         struct lmv_object       *obj;
 215         struct list_head        *cur;
 216
 217         list_for_each(cur, &obj_list) {
 218                 obj = list_entry(cur, struct lmv_object, lo_list);
 219
 220                 /*
 221                  * Check if object is in destroying phase. If so - skip
 222                  * it.
 223                  */
 224                 if (obj->lo_state & O_FREEING)
 225                         continue;
 226
 227                 /*
 228                  * We should make sure, that we have found object belong to
 229                  * passed obd. It is possible that, object manager will have two
 230                  * objects with the same fid belong to different obds, if client
 231                  * and mds runs on the same host. May be it is good idea to have
 232                  * objects list associated with obd.
 233                  */
 234                 if (obj->lo_obd != obd)
 235                         continue;
 236
 237                 /*
 238                  * Check if this is what we're looking for.
 239                  */
 240                 if (lu_fid_eq(&obj->lo_fid, fid))
 241                         return __lmv_object_get(obj);
 242         }
 243
 244         return NULL;
 245 }
 246
 247 struct lmv_object *lmv_object_find(struct obd_device *obd,
 248                                    const struct lu_fid *fid)
 249 {
 250         struct lmv_object       *obj;
 251         ENTRY;
 252
 253         spin_lock(&obj_list_lock);
 254         obj = __lmv_object_find(obd, fid);
 255         spin_unlock(&obj_list_lock);
 256
 257         RETURN(obj);
 258 }
 259
 260 struct lmv_object *lmv_object_find_lock(struct obd_device *obd,
 261                                         const struct lu_fid *fid)
 262 {
 263         struct lmv_object       *obj;
 264         ENTRY;
 265
 266         obj = lmv_object_find(obd, fid);
 267         if (obj)
 268                 lmv_object_lock(obj);
 269
 270         RETURN(obj);
 271 }
 272
 273 static struct lmv_object *__lmv_object_create(struct obd_device *obd,
 274                                               const struct lu_fid *fid,
 275                                               struct lmv_stripe_md *mea)
 276 {
 277         struct lmv_object       *new;
 278         struct lmv_object       *obj;
 279         ENTRY;
 280
 281         obj = lmv_object_find(obd, fid);
 282         if (obj)
 283                 RETURN(obj);
 284
 285         new = lmv_object_alloc(obd, fid, mea);
 286         if (!new)
 287                 RETURN(NULL);
 288
 289         /*
 290          * Check if someone created it already while we were dealing with
 291          * allocating @obj.
 292          */
 293         spin_lock(&obj_list_lock);
 294         obj = __lmv_object_find(obd, fid);
 295         if (obj) {
 296                 /*
 297                  * Someone created it already - put @obj and getting out.
 298                  */
 299                 spin_unlock(&obj_list_lock);
 300                 lmv_object_free(new);
 301                 RETURN(obj);
 302         }
 303
 304         __lmv_object_add(new);
 305         __lmv_object_get(new);
 306
 307         spin_unlock(&obj_list_lock);
 308
 309         CDEBUG(D_INODE, "New obj in lmv cache: "DFID"\n",
 310                PFID(fid));
 311
 312         RETURN(new);
 313 }
 314
 315 struct lmv_object *lmv_object_create(struct obd_export *exp,
 316                                      const struct lu_fid *fid,
 317                                      struct lmv_stripe_md *mea)
 318 {
 319         struct obd_device       *obd = exp->exp_obd;
 320         struct lmv_obd          *lmv = &obd->u.lmv;
 321         struct ptlrpc_request   *req = NULL;
 322         struct lmv_tgt_desc     *tgt;
 323         struct lmv_object       *obj;
 324         struct lustre_md         md;
 325         int                      mealen;
 326         int                      rc;
 327         ENTRY;
 328
 329         CDEBUG(D_INODE, "Get mea for "DFID" and create lmv obj\n",
 330                PFID(fid));
 331
 332         md.mea = NULL;
 333
 334         if (mea == NULL) {
 335                 __u64 valid;
 336
 337                 CDEBUG(D_INODE, "Mea isn't passed in, get it now\n");
 338                 mealen = lmv_get_easize(lmv);
 339
 340                 /*
 341                  * Time to update mea of parent fid.
 342                  */
 343                 md.mea = NULL;
 344                 valid = OBD_MD_FLEASIZE | OBD_MD_FLDIREA | OBD_MD_MEA;
 345
 346                 tgt = lmv_find_target(lmv, fid);
 347                 if (IS_ERR(tgt))
 348                         GOTO(cleanup, obj = (void *)tgt);
 349
 350                 rc = md_getattr(tgt->ltd_exp, fid, NULL, valid, mealen, &req);
 351                 if (rc) {
 352                         CERROR("md_getattr() failed, error %d\n", rc);
 353                         GOTO(cleanup, obj = ERR_PTR(rc));
 354                 }
 355
 356                 rc = md_get_lustre_md(exp, req, NULL, exp, &md);
 357                 if (rc) {
 358                         CERROR("md_get_lustre_md() failed, error %d\n", rc);
 359                         GOTO(cleanup, obj = ERR_PTR(rc));
 360                 }
 361
 362                 if (md.mea == NULL)
 363                         GOTO(cleanup, obj = ERR_PTR(-ENODATA));
 364
 365                 mea = md.mea;
 366         }
 367
 368         /*
 369          * Got mea, now create obj for it.
 370          */
 371         obj = __lmv_object_create(obd, fid, mea);
 372         if (!obj) {
 373                 CERROR("Can't create new object "DFID"\n",
 374                        PFID(fid));
 375                 GOTO(cleanup, obj = ERR_PTR(-ENOMEM));
 376         }
 377
 378         if (md.mea != NULL)
 379                 obd_free_memmd(exp, (void *)&md.mea);
 380
 381         EXIT;
 382 cleanup:
 383         if (req)
 384                 ptlrpc_req_finished(req);
 385         return obj;
 386 }
 387
 388 int lmv_object_delete(struct obd_export *exp, const struct lu_fid *fid)
 389 {
 390         struct obd_device       *obd = exp->exp_obd;
 391         struct lmv_object       *obj;
 392         int                      rc = 0;
 393         ENTRY;
 394
 395         spin_lock(&obj_list_lock);
 396         obj = __lmv_object_find(obd, fid);
 397         if (obj) {
 398                 obj->lo_state |= O_FREEING;
 399                 __lmv_object_put(obj);
 400                 __lmv_object_put(obj);
 401                 rc = 1;
 402         }
 403         spin_unlock(&obj_list_lock);
 404         RETURN(rc);
 405 }
 406
 407 int lmv_object_setup(struct obd_device *obd)
 408 {
 409         ENTRY;
 410         LASSERT(obd != NULL);
 411
 412         CDEBUG(D_INFO, "LMV object manager setup (%s)\n",
 413                obd->obd_uuid.uuid);
 414
 415         RETURN(0);
 416 }
 417
 418 void lmv_object_cleanup(struct obd_device *obd)
 419 {
 420         struct list_head        *cur;
 421         struct list_head        *tmp;
 422         struct lmv_object       *obj;
 423         ENTRY;
 424
 425         CDEBUG(D_INFO, "LMV object manager cleanup (%s)\n",
 426                obd->obd_uuid.uuid);
 427
 428         spin_lock(&obj_list_lock);
 429         list_for_each_safe(cur, tmp, &obj_list) {
 430                 obj = list_entry(cur, struct lmv_object, lo_list);
 431
 432                 if (obj->lo_obd != obd)
 433                         continue;
 434
 435                 obj->lo_state |= O_FREEING;
 436                 if (atomic_read(&obj->lo_count) > 1) {
 437                         CERROR("Object "DFID" has count (%d)\n",
 438                                PFID(&obj->lo_fid), atomic_read(&obj->lo_count));
 439                 }
 440                 __lmv_object_put(obj);
 441         }
 442         spin_unlock(&obj_list_lock);
 443         EXIT;
 444 }