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