lustre/lmv/lmv_objmgr.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 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 <linux/obd_support.h>
  39 #include <linux/lustre_lib.h>
  40 #include <linux/lustre_net.h>
  41 #include <linux/lustre_idl.h>
  42 #include <linux/lustre_dlm.h>
  43 #include <linux/lustre_mds.h>
  44 #include <linux/obd_class.h>
  45 #include <linux/obd_ost.h>
  46 #include <linux/lprocfs_status.h>
  47 #include <linux/lustre_fsfilt.h>
  48 #include <linux/obd_lmv.h>
  49 #include "lmv_internal.h"
  50
  51 static LIST_HEAD(lmv_obj_list);
  52 static spinlock_t lmv_obj_list_lock = SPIN_LOCK_UNLOCKED;
  53
  54 /* creates new obj on passed @fid and @mea. */
  55 struct lmv_obj *
  56 lmv_alloc_obj(struct obd_device *obd, struct ll_fid *fid,
  57               struct mea *mea)
  58 {
  59         int i;
  60         struct lmv_obj *obj;
  61         unsigned int obj_size;
  62         struct lmv_obd *lmv = &obd->u.lmv;
  63
  64         LASSERT(mea->mea_magic == MEA_MAGIC_LAST_CHAR
  65                         || mea->mea_magic == MEA_MAGIC_ALL_CHARS);
  66
  67         OBD_ALLOC(obj, sizeof(*obj));
  68         if (!obj)
  69                 return NULL;
  70
  71         obj->obd = obd;
  72         obj->state = 0;
  73         obj->fid = *fid;
  74         obj->hashtype = mea->mea_magic;
  75
  76         init_MUTEX(&obj->guard);
  77         atomic_set(&obj->count, 0);
  78         obj->objcount = mea->mea_count;
  79
  80         obj_size = sizeof(struct lmv_inode) *
  81                 lmv->desc.ld_tgt_count;
  82
  83         OBD_ALLOC(obj->objs, obj_size);
  84         if (!obj->objs)
  85                 goto err_obj;
  86
  87         memset(obj->objs, 0, obj_size);
  88
  89         /* put all fids in */
  90         for (i = 0; i < mea->mea_count; i++) {
  91                 CDEBUG(D_OTHER, "subobj %lu/%lu/%lu\n",
  92                        (unsigned long)mea->mea_fids[i].mds,
  93                        (unsigned long)mea->mea_fids[i].id,
  94                        (unsigned long)mea->mea_fids[i].generation);
  95                 obj->objs[i].fid = mea->mea_fids[i];
  96         }
  97
  98         return obj;
  99
 100 err_obj:
 101         OBD_FREE(obj, sizeof(*obj));
 102         return NULL;
 103 }
 104
 105 /* destroys passed @obj. */
 106 void
 107 lmv_free_obj(struct lmv_obj *obj)
 108 {
 109         unsigned int obj_size;
 110         struct lmv_obd *lmv = &obj->obd->u.lmv;
 111
 112         obj_size = sizeof(struct lmv_inode) *
 113                 lmv->desc.ld_tgt_count;
 114
 115         OBD_FREE(obj->objs, obj_size);
 116         OBD_FREE(obj, sizeof(*obj));
 117 }
 118
 119 static void
 120 __add_obj(struct lmv_obj *obj)
 121 {
 122         atomic_inc(&obj->count);
 123         list_add(&obj->list, &lmv_obj_list);
 124 }
 125
 126 void
 127 lmv_add_obj(struct lmv_obj *obj)
 128 {
 129         spin_lock(&lmv_obj_list_lock);
 130         __add_obj(obj);
 131         spin_unlock(&lmv_obj_list_lock);
 132 }
 133
 134 static void
 135 __del_obj(struct lmv_obj *obj)
 136 {
 137         if (!(obj->state & O_FREEING))
 138                 LBUG();
 139
 140         list_del(&obj->list);
 141         lmv_free_obj(obj);
 142 }
 143
 144 void
 145 lmv_del_obj(struct lmv_obj *obj)
 146 {
 147         spin_lock(&lmv_obj_list_lock);
 148         __del_obj(obj);
 149         spin_unlock(&lmv_obj_list_lock);
 150 }
 151
 152 static struct lmv_obj *
 153 __get_obj(struct lmv_obj *obj)
 154 {
 155         LASSERT(obj);
 156         atomic_inc(&obj->count);
 157         return obj;
 158 }
 159
 160 struct lmv_obj *
 161 lmv_get_obj(struct lmv_obj *obj)
 162 {
 163         spin_lock(&lmv_obj_list_lock);
 164         __get_obj(obj);
 165         spin_unlock(&lmv_obj_list_lock);
 166
 167         return obj;
 168 }
 169
 170 static void
 171 __put_obj(struct lmv_obj *obj)
 172 {
 173         LASSERT(obj);
 174
 175         if (atomic_dec_and_test(&obj->count)) {
 176                 struct ll_fid *fid = &obj->fid;
 177                 CDEBUG(D_OTHER, "last reference to %lu/%lu/%lu - destroying\n",
 178                        (unsigned long)fid->mds, (unsigned long)fid->id,
 179                        (unsigned long)fid->generation);
 180                 __del_obj(obj);
 181         }
 182 }
 183
 184 void
 185 lmv_put_obj(struct lmv_obj *obj)
 186 {
 187         spin_lock(&lmv_obj_list_lock);
 188         __put_obj(obj);
 189         spin_unlock(&lmv_obj_list_lock);
 190 }
 191
 192 static struct lmv_obj *
 193 __grab_obj(struct obd_device *obd, struct ll_fid *fid)
 194 {
 195         struct lmv_obj *obj;
 196         struct list_head *cur;
 197
 198         list_for_each(cur, &lmv_obj_list) {
 199                 obj = list_entry(cur, struct lmv_obj, list);
 200
 201                 /* check if object is in progress of destroying. If so - skip
 202                  * it. */
 203                 if (obj->state & O_FREEING)
 204                         continue;
 205
 206                 /* check if this is waht we're looking for. */
 207                 if (fid_equal(&obj->fid, fid))
 208                         return __get_obj(obj);
 209         }
 210
 211         return NULL;
 212 }
 213
 214 struct lmv_obj *
 215 lmv_grab_obj(struct obd_device *obd, struct ll_fid *fid)
 216 {
 217         struct lmv_obj *obj;
 218         ENTRY;
 219
 220         spin_lock(&lmv_obj_list_lock);
 221         obj = __grab_obj(obd, fid);
 222         spin_unlock(&lmv_obj_list_lock);
 223
 224         RETURN(obj);
 225 }
 226
 227 /* looks in objects list for an object that matches passed @fid. If it is not
 228  * found -- creates it using passed @mea and puts onto list. */
 229 static struct lmv_obj *
 230 __create_obj(struct obd_device *obd, struct ll_fid *fid, struct mea *mea)
 231 {
 232         struct lmv_obj *new, *obj;
 233         ENTRY;
 234
 235         obj = lmv_grab_obj(obd, fid);
 236         if (obj)
 237                 RETURN(obj);
 238
 239         /* no such object yet, allocate and initialize it. */
 240         new = lmv_alloc_obj(obd, fid, mea);
 241         if (!new)
 242                 RETURN(NULL);
 243
 244         /* check if someone create it already while we were dealing with
 245          * allocating @obj. */
 246         spin_lock(&lmv_obj_list_lock);
 247         obj = __grab_obj(obd, fid);
 248         if (obj) {
 249                 /* someone created it already - put @obj and getting out. */
 250                 lmv_free_obj(new);
 251                 spin_unlock(&lmv_obj_list_lock);
 252                 RETURN(obj);
 253         }
 254
 255         __add_obj(new);
 256         __get_obj(new);
 257
 258         spin_unlock(&lmv_obj_list_lock);
 259
 260         CDEBUG(D_OTHER, "new obj in lmv cache: %lu/%lu/%lu\n",
 261                (unsigned long)fid->mds, (unsigned long)fid->id,
 262                (unsigned long)fid->generation);
 263
 264         RETURN(new);
 265
 266 }
 267
 268 /* creates object from passed @fid and @mea. If @mea is NULL, it will be
 269  * obtained from correct MDT and used for constructing the object. */
 270 struct lmv_obj *
 271 lmv_create_obj(struct obd_export *exp, struct ll_fid *fid, struct mea *mea)
 272 {
 273         struct obd_device *obd = exp->exp_obd;
 274         struct lmv_obd *lmv = &obd->u.lmv;
 275         struct ptlrpc_request *req = NULL;
 276         struct lmv_obj *obj;
 277         struct lustre_md md;
 278         int mealen, i, rc;
 279         ENTRY;
 280
 281         CDEBUG(D_OTHER, "get mea for %lu/%lu/%lu and create lmv obj\n",
 282                (unsigned long)fid->mds, (unsigned long)fid->id,
 283                (unsigned long)fid->generation);
 284
 285         if (!mea) {
 286                 unsigned long valid;
 287
 288                 CDEBUG(D_OTHER, "mea isn't passed in, get it now\n");
 289                 mealen = MEA_SIZE_LMV(lmv);
 290
 291                 /* time to update mea of parent fid */
 292                 i = fid->mds;
 293                 md.mea = NULL;
 294
 295                 valid = OBD_MD_FLEASIZE | OBD_MD_FLDIREA;
 296                 rc = md_getattr(lmv->tgts[fid->mds].ltd_exp, fid,
 297                                 valid, mealen, &req);
 298                 if (rc) {
 299                         CERROR("md_getattr() failed, error %d\n", rc);
 300                         GOTO(cleanup, obj = ERR_PTR(rc));
 301                 }
 302
 303                 rc = mdc_req2lustre_md(exp, req, 0, NULL, &md);
 304                 if (rc) {
 305                         CERROR("mdc_req2lustre_md() failed, error %d\n", rc);
 306                         GOTO(cleanup, obj = ERR_PTR(rc));
 307                 }
 308
 309                 if (!md.mea)
 310                         GOTO(cleanup, obj = ERR_PTR(-ENODATA));
 311
 312                 mea = md.mea;
 313         }
 314
 315         /* got mea, now create obj for it. */
 316         obj = __create_obj(obd, fid, mea);
 317         if (!obj) {
 318                 CERROR("Can't create new object %lu/%lu/%lu\n",
 319                        (unsigned long)fid->mds, (unsigned long)fid->id,
 320                        (unsigned long)fid->generation);
 321                 GOTO(cleanup, obj = ERR_PTR(-ENOMEM));
 322         }
 323 cleanup:
 324         if (req)
 325                 ptlrpc_req_finished(req);
 326         RETURN(obj);
 327 }
 328
 329 /* looks for object with @fid and orders to destroy it. It possible the object
 330  * will not be destroyed right now, because it is still using by someone. In
 331  * this case it will be marked as "freeing" and will not be accessible anymore
 332  * for subsequent callers of lmv_grab_obj(). */
 333 int
 334 lmv_delete_obj(struct obd_export *exp, struct ll_fid *fid)
 335 {
 336         struct obd_device *obd = exp->exp_obd;
 337         struct lmv_obj *obj;
 338         int rc = 0;
 339         ENTRY;
 340
 341         spin_lock(&lmv_obj_list_lock);
 342
 343         obj = __grab_obj(obd, fid);
 344         if (obj) {
 345                 obj->state |= O_FREEING;
 346
 347                 if (atomic_read(&obj->count) > 1)
 348                         CERROR("obj %lu/%lu/%lu has count > 2 (%d)\n",
 349                                (unsigned long) obj->fid.mds,
 350                                (unsigned long) obj->fid.id,
 351                                (unsigned long) obj->fid.generation,
 352                                atomic_read(&obj->count));
 353                 __put_obj(obj);
 354                 __put_obj(obj);
 355                 rc = 1;
 356         }
 357
 358         spin_unlock(&lmv_obj_list_lock);
 359         RETURN(rc);
 360 }
 361
 362 int
 363 lmv_setup_mgr(struct obd_device *obd)
 364 {
 365         CDEBUG(D_INFO, "LMV object manager setup (%s)\n", obd->obd_uuid.uuid);
 366         return 0;
 367 }
 368
 369 void
 370 lmv_cleanup_mgr(struct obd_device *obd)
 371 {
 372         struct lmv_obj *obj;
 373         struct list_head *cur, *tmp;
 374
 375         CDEBUG(D_INFO, "LMV object manager cleanup (%s)\n", obd->obd_uuid.uuid);
 376
 377         spin_lock(&lmv_obj_list_lock);
 378         list_for_each_safe(cur, tmp, &lmv_obj_list) {
 379                 obj = list_entry(cur, struct lmv_obj, list);
 380
 381                 if (obj->obd != obd)
 382                         continue;
 383
 384                 obj->state |= O_FREEING;
 385                 if (atomic_read(&obj->count) > 1)
 386                         CERROR("obj %lu/%lu/%lu has count > 1 (%d)\n",
 387                                (unsigned long) obj->fid.mds,
 388                                (unsigned long) obj->fid.id,
 389                                (unsigned long) obj->fid.generation,
 390                                atomic_read(&obj->count));
 391                 __put_obj(obj);
 392         }
 393         spin_unlock(&lmv_obj_list_lock);
 394 }