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