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