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 CFS_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                 || mea->mea_magic == MEA_MAGIC_HASH_SEGMENT);
  69
  70         OBD_SLAB_ALLOC(obj, obj_cache, CFS_ALLOC_STD,
  71                        sizeof(*obj));
  72         if (!obj)
  73                 return NULL;
  74
  75         atomic_inc(&obj_cache_count);
  76
  77         obj->lo_fid = *fid;
  78         obj->lo_obd = obd;
  79         obj->lo_state = 0;
  80         obj->lo_hashtype = mea->mea_magic;
  81
  82         init_MUTEX(&obj->lo_guard);
  83         atomic_set(&obj->lo_count, 0);
  84         obj->lo_objcount = mea->mea_count;
  85
  86         obj_size = sizeof(struct lmv_inode) *
  87                 lmv->desc.ld_tgt_count;
  88
  89         OBD_ALLOC(obj->lo_inodes, obj_size);
  90         if (!obj->lo_inodes)
  91                 goto err_obj;
  92
  93         memset(obj->lo_inodes, 0, obj_size);
  94
  95         /* put all ids in */
  96         for (i = 0; i < mea->mea_count; i++) {
  97                 int rc;
  98
  99                 CDEBUG(D_OTHER, "subobj "DFID"\n",
 100                        PFID(&mea->mea_ids[i]));
 101                 obj->lo_inodes[i].li_fid = mea->mea_ids[i];
 102                 LASSERT(fid_is_sane(&obj->lo_inodes[i].li_fid));
 103
 104                 /*
 105                  * Cache slave mds number to use it in all cases it is needed
 106                  * instead of constant lookup.
 107                  */
 108                 rc = lmv_fld_lookup(lmv, &obj->lo_inodes[i].li_fid,
 109                                     &obj->lo_inodes[i].li_mds);
 110                 if (rc)
 111                         goto err_obj;
 112         }
 113
 114         return obj;
 115
 116 err_obj:
 117         OBD_FREE(obj, sizeof(*obj));
 118         return NULL;
 119 }
 120
 121 /* destroy passed @obj. */
 122 void
 123 lmv_obj_free(struct lmv_obj *obj)
 124 {
 125         struct lmv_obd *lmv = &obj->lo_obd->u.lmv;
 126         unsigned int obj_size;
 127
 128         LASSERT(!atomic_read(&obj->lo_count));
 129
 130         obj_size = sizeof(struct lmv_inode) *
 131                 lmv->desc.ld_tgt_count;
 132
 133         OBD_FREE(obj->lo_inodes, obj_size);
 134         OBD_SLAB_FREE(obj, obj_cache, sizeof(*obj));
 135         atomic_dec(&obj_cache_count);
 136 }
 137
 138 static void
 139 __lmv_obj_add(struct lmv_obj *obj)
 140 {
 141         atomic_inc(&obj->lo_count);
 142         list_add(&obj->lo_list, &obj_list);
 143 }
 144
 145 void
 146 lmv_obj_add(struct lmv_obj *obj)
 147 {
 148         spin_lock(&obj_list_lock);
 149         __lmv_obj_add(obj);
 150         spin_unlock(&obj_list_lock);
 151 }
 152
 153 static void
 154 __lmv_obj_del(struct lmv_obj *obj)
 155 {
 156         list_del(&obj->lo_list);
 157         lmv_obj_free(obj);
 158 }
 159
 160 void
 161 lmv_obj_del(struct lmv_obj *obj)
 162 {
 163         spin_lock(&obj_list_lock);
 164         __lmv_obj_del(obj);
 165         spin_unlock(&obj_list_lock);
 166 }
 167
 168 static struct lmv_obj *
 169 __lmv_obj_get(struct lmv_obj *obj)
 170 {
 171         LASSERT(obj != NULL);
 172         atomic_inc(&obj->lo_count);
 173         return obj;
 174 }
 175
 176 struct lmv_obj *
 177 lmv_obj_get(struct lmv_obj *obj)
 178 {
 179         spin_lock(&obj_list_lock);
 180         __lmv_obj_get(obj);
 181         spin_unlock(&obj_list_lock);
 182         return obj;
 183 }
 184
 185 static void
 186 __lmv_obj_put(struct lmv_obj *obj)
 187 {
 188         LASSERT(obj);
 189
 190         if (atomic_dec_and_test(&obj->lo_count)) {
 191                 CDEBUG(D_OTHER, "last reference to "DFID" - "
 192                        "destroying\n", PFID(&obj->lo_fid));
 193                 __lmv_obj_del(obj);
 194         }
 195 }
 196
 197 void
 198 lmv_obj_put(struct lmv_obj *obj)
 199 {
 200         spin_lock(&obj_list_lock);
 201         __lmv_obj_put(obj);
 202         spin_unlock(&obj_list_lock);
 203 }
 204
 205 static struct lmv_obj *
 206 __lmv_obj_grab(struct obd_device *obd, const struct lu_fid *fid)
 207 {
 208         struct lmv_obj *obj;
 209         struct list_head *cur;
 210
 211         list_for_each(cur, &obj_list) {
 212                 obj = list_entry(cur, struct lmv_obj, lo_list);
 213
 214                 /* check if object is in progress of destroying. If so - skip
 215                  * it. */
 216                 if (obj->lo_state & O_FREEING)
 217                         continue;
 218
 219                 /*
 220                  * we should make sure, that we have found object belong to
 221                  * passed obd. It is possible that, object manager will have two
 222                  * objects with the same fid belong to different obds, if client
 223                  * and mds runs on the same host. May be it is good idea to have
 224                  * objects list associated with obd.
 225                  */
 226                 if (obj->lo_obd != obd)
 227                         continue;
 228
 229                 /* check if this is what we're looking for. */
 230                 if (lu_fid_eq(&obj->lo_fid, fid))
 231                         return __lmv_obj_get(obj);
 232         }
 233
 234         return NULL;
 235 }
 236
 237 struct lmv_obj *
 238 lmv_obj_grab(struct obd_device *obd, const struct lu_fid *fid)
 239 {
 240         struct lmv_obj *obj;
 241         ENTRY;
 242
 243         spin_lock(&obj_list_lock);
 244         obj = __lmv_obj_grab(obd, fid);
 245         spin_unlock(&obj_list_lock);
 246
 247         RETURN(obj);
 248 }
 249
 250 /* looks in objects list for an object that matches passed @fid. If it is not
 251  * found -- creates it using passed @mea and puts onto list. */
 252 static struct lmv_obj *
 253 __lmv_obj_create(struct obd_device *obd, const struct lu_fid *fid,
 254                  struct lmv_stripe_md *mea)
 255 {
 256         struct lmv_obj *new, *obj;
 257         ENTRY;
 258
 259         obj = lmv_obj_grab(obd, fid);
 260         if (obj)
 261                 RETURN(obj);
 262
 263         /* no such object yet, allocate and initialize it. */
 264         new = lmv_obj_alloc(obd, fid, mea);
 265         if (!new)
 266                 RETURN(NULL);
 267
 268         /* check if someone create it already while we were dealing with
 269          * allocating @obj. */
 270         spin_lock(&obj_list_lock);
 271         obj = __lmv_obj_grab(obd, fid);
 272         if (obj) {
 273                 /* someone created it already - put @obj and getting out. */
 274                 spin_unlock(&obj_list_lock);
 275                 lmv_obj_free(new);
 276                 RETURN(obj);
 277         }
 278
 279         __lmv_obj_add(new);
 280         __lmv_obj_get(new);
 281
 282         spin_unlock(&obj_list_lock);
 283
 284         CDEBUG(D_OTHER, "new obj in lmv cache: "DFID"\n",
 285                PFID(fid));
 286
 287         RETURN(new);
 288
 289 }
 290
 291 /* creates object from passed @fid and @mea. If @mea is NULL, it will be
 292  * obtained from correct MDT and used for constructing the object. */
 293 struct lmv_obj *
 294 lmv_obj_create(struct obd_export *exp, const struct lu_fid *fid,
 295                struct lmv_stripe_md *mea)
 296 {
 297         struct obd_device *obd = exp->exp_obd;
 298         struct lmv_obd *lmv = &obd->u.lmv;
 299         struct ptlrpc_request *req = NULL;
 300         struct obd_export *tgt_exp;
 301         struct lmv_obj *obj;
 302         struct lustre_md md;
 303         int mealen, rc;
 304         ENTRY;
 305
 306         CDEBUG(D_OTHER, "get mea for "DFID" and create lmv obj\n",
 307                PFID(fid));
 308
 309         md.mea = NULL;
 310
 311         if (mea == NULL) {
 312                 __u64 valid;
 313
 314                 CDEBUG(D_OTHER, "mea isn't passed in, get it now\n");
 315                 mealen = lmv_get_easize(lmv);
 316
 317                 /* time to update mea of parent fid */
 318                 md.mea = NULL;
 319                 valid = OBD_MD_FLEASIZE | OBD_MD_FLDIREA | OBD_MD_MEA;
 320
 321                 tgt_exp = lmv_find_export(lmv, fid);
 322                 if (IS_ERR(tgt_exp))
 323                         GOTO(cleanup, obj = (void *)tgt_exp);
 324
 325                 rc = md_getattr(tgt_exp, fid, NULL, valid, mealen, &req);
 326                 if (rc) {
 327                         CERROR("md_getattr() failed, error %d\n", rc);
 328                         GOTO(cleanup, obj = ERR_PTR(rc));
 329                 }
 330
 331                 rc = md_get_lustre_md(exp, req, NULL, exp, &md);
 332                 if (rc) {
 333                         CERROR("mdc_get_lustre_md() failed, error %d\n", rc);
 334                         GOTO(cleanup, obj = ERR_PTR(rc));
 335                 }
 336
 337                 if (md.mea == NULL)
 338                         GOTO(cleanup, obj = ERR_PTR(-ENODATA));
 339
 340                 mea = md.mea;
 341         }
 342
 343         /* got mea, now create obj for it. */
 344         obj = __lmv_obj_create(obd, fid, mea);
 345         if (!obj) {
 346                 CERROR("Can't create new object "DFID"\n",
 347                        PFID(fid));
 348                 GOTO(cleanup, obj = ERR_PTR(-ENOMEM));
 349         }
 350
 351         /* XXX LOV STACKING */
 352         if (md.mea != NULL)
 353                 obd_free_memmd(exp, (void *)&md.mea);
 354
 355         EXIT;
 356 cleanup:
 357         if (req)
 358                 ptlrpc_req_finished(req);
 359         return obj;
 360 }
 361
 362 /*
 363  * looks for object with @fid and orders to destroy it. It is possible the object
 364  * will not be destroyed right now, because it is still using by someone. In
 365  * this case it will be marked as "freeing" and will not be accessible anymore
 366  * for subsequent callers of lmv_obj_grab().
 367  */
 368 int
 369 lmv_obj_delete(struct obd_export *exp, const struct lu_fid *fid)
 370 {
 371         struct obd_device *obd = exp->exp_obd;
 372         struct lmv_obj *obj;
 373         int rc = 0;
 374         ENTRY;
 375
 376         spin_lock(&obj_list_lock);
 377         obj = __lmv_obj_grab(obd, fid);
 378         if (obj) {
 379                 obj->lo_state |= O_FREEING;
 380                 __lmv_obj_put(obj);
 381                 __lmv_obj_put(obj);
 382                 rc = 1;
 383         }
 384         spin_unlock(&obj_list_lock);
 385
 386         RETURN(rc);
 387 }
 388
 389 int
 390 lmv_obj_setup(struct obd_device *obd)
 391 {
 392         ENTRY;
 393         LASSERT(obd != NULL);
 394
 395         CDEBUG(D_INFO, "LMV object manager setup (%s)\n",
 396                obd->obd_uuid.uuid);
 397
 398         RETURN(0);
 399 }
 400
 401 void
 402 lmv_obj_cleanup(struct obd_device *obd)
 403 {
 404         struct list_head *cur, *tmp;
 405         struct lmv_obj *obj;
 406         ENTRY;
 407
 408         CDEBUG(D_INFO, "LMV object manager cleanup (%s)\n",
 409                obd->obd_uuid.uuid);
 410
 411         spin_lock(&obj_list_lock);
 412         list_for_each_safe(cur, tmp, &obj_list) {
 413                 obj = list_entry(cur, struct lmv_obj, lo_list);
 414
 415                 if (obj->lo_obd != obd)
 416                         continue;
 417
 418                 obj->lo_state |= O_FREEING;
 419                 if (atomic_read(&obj->lo_count) > 1) {
 420                         CERROR("obj "DFID" has count > 1 (%d)\n",
 421                                PFID(&obj->lo_fid), atomic_read(&obj->lo_count));
 422                 }
 423                 __lmv_obj_put(obj);
 424         }
 425         spin_unlock(&obj_list_lock);
 426         EXIT;
 427 }