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