X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Flmv%2Flmv_objmgr.c;h=ce054c74721c0db1ad2e6833f47e1e36cbdd0e68;hb=95b0089f355d73a695c209d4126b9cc4f1d60c9c;hp=1b1c9cd1b1016a6492369ea1bcd15b4ceb861e45;hpb=e44e9b278432a1df83482b1cd83b2081fabe94dc;p=fs%2Flustre-release.git

diff --git a/lustre/lmv/lmv_objmgr.c b/lustre/lmv/lmv_objmgr.c
index 1b1c9cd..ce054c7 100644
--- a/lustre/lmv/lmv_objmgr.c
+++ b/lustre/lmv/lmv_objmgr.c
@@ -48,168 +48,361 @@
 #include <linux/obd_lmv.h>
 #include "lmv_internal.h"
 
+/* objects cache. */
+extern kmem_cache_t *obj_cache;
+extern atomic_t obj_cache_count;
 
-LIST_HEAD(lmv_obj_list);
-spinlock_t lmv_obj_list_lock = SPIN_LOCK_UNLOCKED;
+/* object list and its guard. */
+static LIST_HEAD(obj_list);
+static spinlock_t obj_list_lock = SPIN_LOCK_UNLOCKED;
 
-struct lmv_obj *lmv_grab_obj(struct obd_device *obd,
-                             struct ll_fid *fid, int create)
+/* creates new obj on passed @id and @mea. */
+struct lmv_obj *
+lmv_alloc_obj(struct obd_device *obd,
+              struct lustre_id *id,
+              struct mea *mea)
 {
+        int i;
+        struct lmv_obj *obj;
+        unsigned int obj_size;
         struct lmv_obd *lmv = &obd->u.lmv;
-        struct list_head *cur;
-        struct lmv_obj *obj, *obj2;
-
-        spin_lock(&lmv_obj_list_lock);
-        list_for_each(cur, &lmv_obj_list) {
-                obj = list_entry(cur, struct lmv_obj, list);
-                if (obj->fid.mds == fid->mds && obj->fid.id == fid->id &&
-                                obj->fid.generation == fid->generation) {
-                        atomic_inc(&obj->count);
-                        spin_unlock(&lmv_obj_list_lock);
-                        RETURN(obj);
-                }
-        }
-        spin_unlock(&lmv_obj_list_lock);
 
-        if (!create)
-                RETURN(NULL);
+        LASSERT(mea->mea_magic == MEA_MAGIC_LAST_CHAR
+                || mea->mea_magic == MEA_MAGIC_ALL_CHARS);
 
-        /* no such object yet, allocate and initialize them */
-        OBD_ALLOC(obj, sizeof(*obj));
+        OBD_SLAB_ALLOC(obj, obj_cache, GFP_NOFS,
+                       sizeof(*obj));
         if (!obj)
-                RETURN(NULL);
-        atomic_set(&obj->count, 0);
-        obj->fid = *fid;
+                return NULL;
+
+        atomic_inc(&obj_cache_count);
+        
+        obj->id = *id;
         obj->obd = obd;
+        obj->state = 0;
+        obj->hashtype = mea->mea_magic;
 
-        OBD_ALLOC(obj->objs, sizeof(struct lmv_inode) * lmv->desc.ld_tgt_count);
-        if (!obj->objs) {
-                OBD_FREE(obj, sizeof(*obj));
-                RETURN(NULL);
-        }
-        memset(obj->objs, 0,  sizeof(struct lmv_inode) * lmv->desc.ld_tgt_count);
-
-        spin_lock(&lmv_obj_list_lock);
-        list_for_each(cur, &lmv_obj_list) {
-                obj2 = list_entry(cur, struct lmv_obj, list);
-                if (obj2->fid.mds == fid->mds && obj2->fid.id == fid->id &&
-                                obj2->fid.generation == fid->generation) {
-                        /* someone created it already */
-                        OBD_FREE(obj->objs,
-                                  sizeof(struct lmv_inode) * lmv->desc.ld_tgt_count);
-                        OBD_FREE(obj, sizeof(*obj));
-
-                        atomic_inc(&obj2->count);
-                        spin_unlock(&lmv_obj_list_lock);
-                        RETURN(obj2);
-                }
+        init_MUTEX(&obj->guard);
+        atomic_set(&obj->count, 0);
+        obj->objcount = mea->mea_count;
+
+        obj_size = sizeof(struct lmv_inode) *
+                lmv->desc.ld_tgt_count;
+        
+        OBD_ALLOC(obj->objs, obj_size);
+        if (!obj->objs)
+                goto err_obj;
+
+        memset(obj->objs, 0, obj_size);
+
+        /* put all ids in */
+        for (i = 0; i < mea->mea_count; i++) {
+                CDEBUG(D_OTHER, "subobj "DLID4"\n",
+                       OLID4(&mea->mea_ids[i]));
+                obj->objs[i].id = mea->mea_ids[i];
+                LASSERT(id_ino(&obj->objs[i].id));
+                LASSERT(id_fid(&obj->objs[i].id));
         }
-        list_add(&obj->list, &lmv_obj_list);
-        CDEBUG(D_OTHER, "new obj in lmv cache: %lu/%lu/%lu\n",
-               (unsigned long) fid->mds, (unsigned long) fid->id,
-               (unsigned long) fid->generation);
-        spin_unlock(&lmv_obj_list_lock);
 
-        RETURN(obj);
+        return obj;
         
+err_obj:
+        OBD_FREE(obj, sizeof(*obj));
+        return NULL;
 }
 
-void lmv_put_obj(struct lmv_obj *obj)
+/* destroy passed @obj. */
+void
+lmv_free_obj(struct lmv_obj *obj)
 {
-        if (!obj)
-                return;
+        unsigned int obj_size;
+        struct lmv_obd *lmv = &obj->obd->u.lmv;
+        
+        LASSERT(!atomic_read(&obj->count));
+        
+        obj_size = sizeof(struct lmv_inode) *
+                lmv->desc.ld_tgt_count;
+        
+        OBD_FREE(obj->objs, obj_size);
+        OBD_SLAB_FREE(obj, obj_cache, sizeof(*obj));
+        atomic_dec(&obj_cache_count);
+}
+
+static void
+__add_obj(struct lmv_obj *obj)
+{
+        atomic_inc(&obj->count);
+        list_add(&obj->list, &obj_list);
+}
+
+void
+lmv_add_obj(struct lmv_obj *obj)
+{
+        spin_lock(&obj_list_lock);
+        __add_obj(obj);
+        spin_unlock(&obj_list_lock);
+}
+
+static void
+__del_obj(struct lmv_obj *obj)
+{
+        list_del(&obj->list);
+        lmv_free_obj(obj);
+}
+
+void
+lmv_del_obj(struct lmv_obj *obj)
+{
+        spin_lock(&obj_list_lock);
+        __del_obj(obj);
+        spin_unlock(&obj_list_lock);
+}
+
+static struct lmv_obj *
+__get_obj(struct lmv_obj *obj)
+{
+        LASSERT(obj != NULL);
+        atomic_inc(&obj->count);
+        return obj;
+}
+
+struct lmv_obj *
+lmv_get_obj(struct lmv_obj *obj)
+{
+        spin_lock(&obj_list_lock);
+        __get_obj(obj);
+        spin_unlock(&obj_list_lock);
+        return obj;
+}
+
+static void
+__put_obj(struct lmv_obj *obj)
+{
+        LASSERT(obj);
+
         if (atomic_dec_and_test(&obj->count)) {
-                CDEBUG(D_OTHER, "last reference to %lu/%lu/%lu\n",
-                       (unsigned long) obj->fid.mds,
-                       (unsigned long) obj->fid.id,
-                       (unsigned long) obj->fid.generation);
+                struct lustre_id *id = &obj->id;
+                CDEBUG(D_OTHER, "last reference to "DLID4" - "
+                       "destroying\n", OLID4(id));
+                __del_obj(obj);
         }
 }
 
-void lmv_cleanup_objs(struct obd_device *obd)
+void
+lmv_put_obj(struct lmv_obj *obj)
+{
+        spin_lock(&obj_list_lock);
+        __put_obj(obj);
+        spin_unlock(&obj_list_lock);
+}
+
+static struct lmv_obj *
+__grab_obj(struct obd_device *obd, struct lustre_id *id)
 {
-        struct lmv_obd *lmv = &obd->u.lmv;
-        struct list_head *cur, *tmp;
         struct lmv_obj *obj;
+        struct list_head *cur;
 
-        spin_lock(&lmv_obj_list_lock);
-        list_for_each_safe(cur, tmp, &lmv_obj_list) {
+        list_for_each(cur, &obj_list) {
                 obj = list_entry(cur, struct lmv_obj, list);
+
+                /* check if object is in progress of destroying. If so - skip
+                 * it. */
+                if (obj->state & O_FREEING)
+                        continue;
+
+                /* 
+                 * we should make sure, that we have found object belong to
+                 * passed obd. It is possible that, object manager will have two
+                 * objects with the same fid belong to different obds, if client
+                 * and mds runs on the same host. May be it is good idea to have
+                 * objects list assosiated with obd.
+                 */
                 if (obj->obd != obd)
                         continue;
 
-                list_del(&obj->list);
-                OBD_FREE(obj->objs,
-                         sizeof(struct lmv_inode) * lmv->desc.ld_tgt_count);
-                OBD_FREE(obj, sizeof(*obj));
+                /* check if this is what we're looking for. */
+                if (id_equal_fid(&obj->id, id))
+                        return __get_obj(obj);
+        }
+
+        return NULL;
+}
+
+struct lmv_obj *
+lmv_grab_obj(struct obd_device *obd, struct lustre_id *id)
+{
+        struct lmv_obj *obj;
+        ENTRY;
+        
+        spin_lock(&obj_list_lock);
+        obj = __grab_obj(obd, id);
+        spin_unlock(&obj_list_lock);
+        
+        RETURN(obj);
+}
+
+/* looks in objects list for an object that matches passed @id. If it is not
+ * found -- creates it using passed @mea and puts onto list. */
+static struct lmv_obj *
+__create_obj(struct obd_device *obd, struct lustre_id *id, struct mea *mea)
+{
+        struct lmv_obj *new, *obj;
+        ENTRY;
+
+        obj = lmv_grab_obj(obd, id);
+        if (obj)
+                RETURN(obj);
+
+        /* no such object yet, allocate and initialize it. */
+        new = lmv_alloc_obj(obd, id, mea);
+        if (!new)
+                RETURN(NULL);
+
+        /* check if someone create it already while we were dealing with
+         * allocating @obj. */
+        spin_lock(&obj_list_lock);
+        obj = __grab_obj(obd, id);
+        if (obj) {
+                /* someone created it already - put @obj and getting out. */
+                lmv_free_obj(new);
+                spin_unlock(&obj_list_lock);
+                RETURN(obj);
         }
-        spin_unlock(&lmv_obj_list_lock);
+
+        __add_obj(new);
+        __get_obj(new);
+        
+        spin_unlock(&obj_list_lock);
+
+        CDEBUG(D_OTHER, "new obj in lmv cache: "DLID4"\n",
+               OLID4(id));
+
+        RETURN(new);
+        
 }
 
-int lmv_create_obj_from_attrs(struct obd_export *exp,
-                              struct ll_fid *fid, struct mea *mea)
+/* creates object from passed @id and @mea. If @mea is NULL, it will be
+ * obtained from correct MDT and used for constructing the object. */
+struct lmv_obj *
+lmv_create_obj(struct obd_export *exp, struct lustre_id *id, struct mea *mea)
 {
         struct obd_device *obd = exp->exp_obd;
         struct lmv_obd *lmv = &obd->u.lmv;
         struct ptlrpc_request *req = NULL;
         struct lmv_obj *obj;
         struct lustre_md md;
-        int mealen, i, rc = 0;
+        int mealen, rc;
         ENTRY;
 
-        CDEBUG(D_OTHER, "get mea for %lu/%lu/%lu and create lmv obj\n",
-               (unsigned long) fid->mds, (unsigned long) fid->id,
-               (unsigned long) fid->generation);
+        CDEBUG(D_OTHER, "get mea for "DLID4" and create lmv obj\n",
+               OLID4(id));
 
         if (!mea) {
-                unsigned long valid;
+                __u64 valid;
                 
                 CDEBUG(D_OTHER, "mea isn't passed in, get it now\n");
                 mealen = MEA_SIZE_LMV(lmv);
                 
-                /* time to update mea of parent fid */
-                i = fid->mds;
+                /* time to update mea of parent id */
                 md.mea = NULL;
-                
                 valid = OBD_MD_FLEASIZE | OBD_MD_FLDIREA;
-                rc = md_getattr(lmv->tgts[fid->mds].ltd_exp, fid,
-                                valid, mealen, &req);
+
+                rc = md_getattr(lmv->tgts[id_group(id)].ltd_exp,
+                                id, valid, mealen, &req);
                 if (rc) {
-                        CERROR("md_getattr() failed, rc = %d\n", rc);
-                        GOTO(cleanup, rc);
+                        CERROR("md_getattr() failed, error %d\n", rc);
+                        GOTO(cleanup, obj = ERR_PTR(rc));
                 }
 
                 rc = mdc_req2lustre_md(exp, req, 0, NULL, &md);
                 if (rc) {
-                        CERROR("mdc_req2lustre_md() failed, rc = %d\n", rc);
-                        GOTO(cleanup, rc);
+                        CERROR("mdc_req2lustre_md() failed, error %d\n", rc);
+                        GOTO(cleanup, obj = ERR_PTR(rc));
                 }
 
-                if (md.mea == NULL)
-                        GOTO(cleanup, rc = -ENODATA);
+                if (!md.mea)
+                        GOTO(cleanup, obj = ERR_PTR(-ENODATA));
                         
                 mea = md.mea;
         }
 
-        /* got mea, now create obj for it */
-        obj = lmv_grab_obj(obd, fid, 1);
-        if (!obj)
-                GOTO(cleanup, rc = -ENOMEM);
-
-        obj->objcount = mea->mea_count;
-        /* put all fids in */
-        for (i = 0; i < mea->mea_count; i++) {
-                CDEBUG(D_OTHER, "subobj %lu/%lu/%lu\n",
-                       (unsigned long) mea->mea_fids[i].mds,
-                       (unsigned long) mea->mea_fids[i].id,
-                       (unsigned long) mea->mea_fids[i].generation);
-                obj->objs[i].fid = mea->mea_fids[i];
+        /* got mea, now create obj for it. */
+        obj = __create_obj(obd, id, mea);
+        if (!obj) {
+                CERROR("Can't create new object "DLID4"\n",
+                       OLID4(id));
+                GOTO(cleanup, obj = ERR_PTR(-ENOMEM));
         }
-
+        EXIT;
 cleanup:
-        if (req)       
+        if (req)
                 ptlrpc_req_finished(req);
-        RETURN(rc); 
+        return obj;
 }
 
+/*
+ * looks for object with @id and orders to destroy it. It is possible the object
+ * will not be destroyed right now, because it is still using by someone. In
+ * this case it will be marked as "freeing" and will not be accessible anymore
+ * for subsequent callers of lmv_grab_obj().
+ */
+int
+lmv_delete_obj(struct obd_export *exp, struct lustre_id *id)
+{
+        struct obd_device *obd = exp->exp_obd;
+        struct lmv_obj *obj;
+        int rc = 0;
+        ENTRY;
+
+        spin_lock(&obj_list_lock);
+        obj = __grab_obj(obd, id);
+        if (obj) {
+                obj->state |= O_FREEING;
+                __put_obj(obj);
+                __put_obj(obj);
+                rc = 1;
+        }
+        spin_unlock(&obj_list_lock);
+
+        RETURN(rc);
+}
+
+int
+lmv_setup_mgr(struct obd_device *obd)
+{
+        ENTRY;
+        LASSERT(obd != NULL);
+        
+        CDEBUG(D_INFO, "LMV object manager setup (%s)\n",
+               obd->obd_uuid.uuid);
+
+        RETURN(0);
+}
 
+void
+lmv_cleanup_mgr(struct obd_device *obd)
+{
+        struct list_head *cur, *tmp;
+        struct lmv_obj *obj;
+        ENTRY;
+
+        CDEBUG(D_INFO, "LMV object manager cleanup (%s)\n",
+               obd->obd_uuid.uuid);
+        
+        spin_lock(&obj_list_lock);
+        list_for_each_safe(cur, tmp, &obj_list) {
+                obj = list_entry(cur, struct lmv_obj, list);
+                
+                if (obj->obd != obd)
+                        continue;
+
+                obj->state |= O_FREEING;
+                if (atomic_read(&obj->count) > 1) {
+                        CERROR("obj "DLID4" has count > 1 (%d)\n",
+                               OLID4(&obj->id), atomic_read(&obj->count));
+                }
+                __put_obj(obj);
+        }
+        spin_unlock(&obj_list_lock);
+        EXIT;
+}