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  * Copyright (C) 2002, 2003, 2004, 2005, 2006 Cluster File Systems, Inc.
   5  *
   6  *   This file is part of Lustre, http://www.lustre.org.
   7  *
   8  *   Lustre is free software; you can redistribute it and/or
   9  *   modify it under the terms of version 2 of the GNU General Public
  10  *   License as published by the Free Software Foundation.
  11  *
  12  *   Lustre is distributed in the hope that it will be useful,
  13  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  *   GNU General Public License for more details.
  16  *
  17  *   You should have received a copy of the GNU General Public License
  18  *   along with Lustre; if not, write to the Free Software
  19  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  20  */
  21
  22 #ifndef EXPORT_SYMTAB
  23 # define EXPORT_SYMTAB
  24 #endif
  25 #define DEBUG_SUBSYSTEM S_LMV
  26 #ifdef __KERNEL__
  27 #include <linux/slab.h>
  28 #include <linux/module.h>
  29 #include <linux/init.h>
  30 #include <linux/slab.h>
  31 #include <linux/pagemap.h>
  32 #include <asm/div64.h>
  33 #include <linux/seq_file.h>
  34 #else
  35 #include <liblustre.h>
  36 #endif
  37
  38 #include <lustre/lustre_idl.h>
  39 #include <obd_support.h>
  40 #include <lustre_lib.h>
  41 #include <lustre_net.h>
  42 #include <lustre_dlm.h>
  43 #include <obd_class.h>
  44 #include <lprocfs_status.h>
  45 #include "lmv_internal.h"
  46
  47 /* objects cache. */
  48 extern cfs_mem_cache_t *obj_cache;
  49 extern atomic_t obj_cache_count;
  50
  51 /* object list and its guard. */
  52 static LIST_HEAD(obj_list);
  53 static spinlock_t obj_list_lock = SPIN_LOCK_UNLOCKED;
  54
  55 /* creates new obj on passed @fid and @mea. */
  56 struct lmv_obj *
  57 lmv_obj_alloc(struct obd_device *obd,
  58               const struct lu_fid *fid,
  59               struct lmv_stripe_md *mea)
  60 {
  61         int i;
  62         struct lmv_obj *obj;
  63         unsigned int obj_size;
  64         struct lmv_obd *lmv = &obd->u.lmv;
  65
  66         LASSERT(mea->mea_magic == MEA_MAGIC_LAST_CHAR
  67                 || mea->mea_magic == MEA_MAGIC_ALL_CHARS);
  68
  69         OBD_SLAB_ALLOC(obj, obj_cache, CFS_ALLOC_STD,
  70                        sizeof(*obj));
  71         if (!obj)
  72                 return NULL;
  73
  74         atomic_inc(&obj_cache_count);
  75
  76         obj->lo_fid = *fid;
  77         obj->lo_obd = obd;
  78         obj->lo_state = 0;
  79         obj->lo_hashtype = mea->mea_magic;
  80
  81         init_MUTEX(&obj->lo_guard);
  82         atomic_set(&obj->lo_count, 0);
  83         obj->lo_objcount = mea->mea_count;
  84
  85         obj_size = sizeof(struct lmv_inode) *
  86                 lmv->desc.ld_tgt_count;
  87
  88         OBD_ALLOC(obj->lo_inodes, obj_size);
  89         if (!obj->lo_inodes)
  90                 goto err_obj;
  91
  92         memset(obj->lo_inodes, 0, obj_size);
  93
  94         /* put all ids in */
  95         for (i = 0; i < mea->mea_count; i++) {
  96                 CDEBUG(D_OTHER, "subobj "DFID"\n",
  97                        PFID(&mea->mea_ids[i]));
  98                 obj->lo_inodes[i].li_fid = mea->mea_ids[i];
  99                 LASSERT(fid_is_sane(&obj->lo_inodes[i].li_fid));
 100         }
 101
 102         return obj;
 103
 104 err_obj:
 105         OBD_FREE(obj, sizeof(*obj));
 106         return NULL;
 107 }
 108
 109 /* destroy passed @obj. */
 110 void
 111 lmv_obj_free(struct lmv_obj *obj)
 112 {
 113         struct lmv_obd *lmv = &obj->lo_obd->u.lmv;
 114         unsigned int obj_size;
 115
 116         LASSERT(!atomic_read(&obj->lo_count));
 117
 118         obj_size = sizeof(struct lmv_inode) *
 119                 lmv->desc.ld_tgt_count;
 120
 121         OBD_FREE(obj->lo_inodes, obj_size);
 122         OBD_SLAB_FREE(obj, obj_cache, sizeof(*obj));
 123         atomic_dec(&obj_cache_count);
 124 }
 125
 126 static void
 127 __lmv_obj_add(struct lmv_obj *obj)
 128 {
 129         atomic_inc(&obj->lo_count);
 130         list_add(&obj->lo_list, &obj_list);
 131 }
 132
 133 void
 134 lmv_obj_add(struct lmv_obj *obj)
 135 {
 136         spin_lock(&obj_list_lock);
 137         __lmv_obj_add(obj);
 138         spin_unlock(&obj_list_lock);
 139 }
 140
 141 static void
 142 __lmv_obj_del(struct lmv_obj *obj)
 143 {
 144         list_del(&obj->lo_list);
 145         lmv_obj_free(obj);
 146 }
 147
 148 void
 149 lmv_obj_del(struct lmv_obj *obj)
 150 {
 151         spin_lock(&obj_list_lock);
 152         __lmv_obj_del(obj);
 153         spin_unlock(&obj_list_lock);
 154 }
 155
 156 static struct lmv_obj *
 157 __lmv_obj_get(struct lmv_obj *obj)
 158 {
 159         LASSERT(obj != NULL);
 160         atomic_inc(&obj->lo_count);
 161         return obj;
 162 }
 163
 164 struct lmv_obj *
 165 lmv_obj_get(struct lmv_obj *obj)
 166 {
 167         spin_lock(&obj_list_lock);
 168         __lmv_obj_get(obj);
 169         spin_unlock(&obj_list_lock);
 170         return obj;
 171 }
 172
 173 static void
 174 __lmv_obj_put(struct lmv_obj *obj)
 175 {
 176         LASSERT(obj);
 177
 178         if (atomic_dec_and_test(&obj->lo_count)) {
 179                 struct lu_fid *fid = &obj->lo_fid;
 180                 CDEBUG(D_OTHER, "last reference to "DFID" - "
 181                        "destroying\n", PFID(fid));
 182                 __lmv_obj_del(obj);
 183         }
 184 }
 185
 186 void
 187 lmv_obj_put(struct lmv_obj *obj)
 188 {
 189         spin_lock(&obj_list_lock);
 190         __lmv_obj_put(obj);
 191         spin_unlock(&obj_list_lock);
 192 }
 193
 194 static struct lmv_obj *
 195 __lmv_obj_grab(struct obd_device *obd, const struct lu_fid *fid)
 196 {
 197         struct lmv_obj *obj;
 198         struct list_head *cur;
 199
 200         list_for_each(cur, &obj_list) {
 201                 obj = list_entry(cur, struct lmv_obj, lo_list);
 202
 203                 /* check if object is in progress of destroying. If so - skip
 204                  * it. */
 205                 if (obj->lo_state & O_FREEING)
 206                         continue;
 207
 208                 /*
 209                  * we should make sure, that we have found object belong to
 210                  * passed obd. It is possible that, object manager will have two
 211                  * objects with the same fid belong to different obds, if client
 212                  * and mds runs on the same host. May be it is good idea to have
 213                  * objects list associated with obd.
 214                  */
 215                 if (obj->lo_obd != obd)
 216                         continue;
 217
 218                 /* check if this is what we're looking for. */
 219                 if (lu_fid_eq(&obj->lo_fid, fid))
 220                         return __lmv_obj_get(obj);
 221         }
 222
 223         return NULL;
 224 }
 225
 226 struct lmv_obj *
 227 lmv_obj_grab(struct obd_device *obd, const struct lu_fid *fid)
 228 {
 229         struct lmv_obj *obj;
 230         ENTRY;
 231
 232         spin_lock(&obj_list_lock);
 233         obj = __lmv_obj_grab(obd, fid);
 234         spin_unlock(&obj_list_lock);
 235
 236         RETURN(obj);
 237 }
 238
 239 /* looks in objects list for an object that matches passed @fid. If it is not
 240  * found -- creates it using passed @mea and puts onto list. */
 241 static struct lmv_obj *
 242 __lmv_obj_create(struct obd_device *obd, const struct lu_fid *fid,
 243                  struct lmv_stripe_md *mea)
 244 {
 245         struct lmv_obj *new, *obj;
 246         ENTRY;
 247
 248         obj = lmv_obj_grab(obd, fid);
 249         if (obj)
 250                 RETURN(obj);
 251
 252         /* no such object yet, allocate and initialize it. */
 253         new = lmv_obj_alloc(obd, fid, mea);
 254         if (!new)
 255                 RETURN(NULL);
 256
 257         /* check if someone create it already while we were dealing with
 258          * allocating @obj. */
 259         spin_lock(&obj_list_lock);
 260         obj = __lmv_obj_grab(obd, fid);
 261         if (obj) {
 262                 /* someone created it already - put @obj and getting out. */
 263                 spin_unlock(&obj_list_lock);
 264                 lmv_obj_free(new);
 265                 RETURN(obj);
 266         }
 267
 268         __lmv_obj_add(new);
 269         __lmv_obj_get(new);
 270
 271         spin_unlock(&obj_list_lock);
 272
 273         CDEBUG(D_OTHER, "new obj in lmv cache: "DFID"\n",
 274                PFID(fid));
 275
 276         RETURN(new);
 277
 278 }
 279
 280 /* creates object from passed @fid and @mea. If @mea is NULL, it will be
 281  * obtained from correct MDT and used for constructing the object. */
 282 struct lmv_obj *
 283 lmv_obj_create(struct obd_export *exp, const struct lu_fid *fid,
 284                struct lmv_stripe_md *mea)
 285 {
 286         struct obd_device *obd = exp->exp_obd;
 287         struct lmv_obd *lmv = &obd->u.lmv;
 288         struct ptlrpc_request *req = NULL;
 289         struct obd_export *tgt_exp;
 290         struct lmv_obj *obj;
 291         struct lustre_md md;
 292         int mealen, rc;
 293         ENTRY;
 294
 295         CDEBUG(D_OTHER, "get mea for "DFID" and create lmv obj\n",
 296                PFID(fid));
 297
 298         md.mea = NULL;
 299
 300         if (mea == NULL) {
 301                 __u64 valid;
 302
 303                 CDEBUG(D_OTHER, "mea isn't passed in, get it now\n");
 304                 mealen = lmv_get_easize(lmv);
 305
 306                 /* time to update mea of parent fid */
 307                 md.mea = NULL;
 308                 valid = OBD_MD_FLEASIZE | OBD_MD_FLDIREA | OBD_MD_MEA;
 309
 310                 tgt_exp = lmv_get_export(lmv, fid);
 311                 if (IS_ERR(tgt_exp))
 312                         GOTO(cleanup, obj = (void *)tgt_exp);
 313
 314                 rc = md_getattr(tgt_exp, fid, valid, mealen, &req);
 315                 if (rc) {
 316                         CERROR("md_getattr() failed, error %d\n", rc);
 317                         GOTO(cleanup, obj = ERR_PTR(rc));
 318                 }
 319
 320                 rc = md_get_lustre_md(exp, req, 0, NULL, &md);
 321                 if (rc) {
 322                         CERROR("mdc_get_lustre_md() failed, error %d\n", rc);
 323                         GOTO(cleanup, obj = ERR_PTR(rc));
 324                 }
 325
 326                 if (md.mea == NULL)
 327                         GOTO(cleanup, obj = ERR_PTR(-ENODATA));
 328
 329                 mea = md.mea;
 330         }
 331
 332         /* got mea, now create obj for it. */
 333         obj = __lmv_obj_create(obd, fid, mea);
 334         if (!obj) {
 335                 CERROR("Can't create new object "DFID"\n",
 336                        PFID(fid));
 337                 GOTO(cleanup, obj = ERR_PTR(-ENOMEM));
 338         }
 339
 340         if (md.mea != NULL)
 341                 obd_free_memmd(exp, (struct lov_stripe_md **)&md.mea);
 342
 343         EXIT;
 344 cleanup:
 345         if (req)
 346                 ptlrpc_req_finished(req);
 347         return obj;
 348 }
 349
 350 /*
 351  * looks for object with @fid and orders to destroy it. It is possible the object
 352  * will not be destroyed right now, because it is still using by someone. In
 353  * this case it will be marked as "freeing" and will not be accessible anymore
 354  * for subsequent callers of lmv_obj_grab().
 355  */
 356 int
 357 lmv_obj_delete(struct obd_export *exp, const struct lu_fid *fid)
 358 {
 359         struct obd_device *obd = exp->exp_obd;
 360         struct lmv_obj *obj;
 361         int rc = 0;
 362         ENTRY;
 363
 364         spin_lock(&obj_list_lock);
 365         obj = __lmv_obj_grab(obd, fid);
 366         if (obj) {
 367                 obj->lo_state |= O_FREEING;
 368                 __lmv_obj_put(obj);
 369                 __lmv_obj_put(obj);
 370                 rc = 1;
 371         }
 372         spin_unlock(&obj_list_lock);
 373
 374         RETURN(rc);
 375 }
 376
 377 int
 378 lmv_obj_setup(struct obd_device *obd)
 379 {
 380         ENTRY;
 381         LASSERT(obd != NULL);
 382
 383         CDEBUG(D_INFO, "LMV object manager setup (%s)\n",
 384                obd->obd_uuid.uuid);
 385
 386         RETURN(0);
 387 }
 388
 389 void
 390 lmv_obj_cleanup(struct obd_device *obd)
 391 {
 392         struct list_head *cur, *tmp;
 393         struct lmv_obj *obj;
 394         ENTRY;
 395
 396         CDEBUG(D_INFO, "LMV object manager cleanup (%s)\n",
 397                obd->obd_uuid.uuid);
 398
 399         spin_lock(&obj_list_lock);
 400         list_for_each_safe(cur, tmp, &obj_list) {
 401                 obj = list_entry(cur, struct lmv_obj, lo_list);
 402
 403                 if (obj->lo_obd != obd)
 404                         continue;
 405
 406                 obj->lo_state |= O_FREEING;
 407                 if (atomic_read(&obj->lo_count) > 1) {
 408                         CERROR("obj "DFID" has count > 1 (%d)\n",
 409                                PFID(&obj->lo_fid), atomic_read(&obj->lo_count));
 410                 }
 411                 __lmv_obj_put(obj);
 412         }
 413         spin_unlock(&obj_list_lock);
 414         EXIT;
 415 }