mdd_device: adjust to dt_object, and some cleanups

diff --git a/lustre/mdd/mdd_handler.c b/lustre/mdd/mdd_handler.c
index 37cd7c1..d7de9bb 100644 (file)
--- a/lustre/mdd/mdd_handler.c
+++ b/lustre/mdd/mdd_handler.c
@@ -1,4 +1,4 @@
-/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
+/* -*- MODE: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
  * vim:expandtab:shiftwidth=8:tabstop=8:
  *
  *  mdd/mdd_handler.c
@@ -45,7 +45,12 @@
 #include "mdd_internal.h"
 
 
+static struct thandle* mdd_trans_start(struct mdd_device *, struct txn_param *);
+static void mdd_trans_stop(struct mdd_device *mdd, struct thandle *handle);
+static struct dt_object* mdd_object_child(struct mdd_object *o);
 static struct lu_device_operations mdd_lu_ops;
+static void mdd_lock(struct mdd_object *obj, enum dt_lock_mode mode);
+static void mdd_unlock(struct mdd_object *obj, enum dt_lock_mode mode);
 
 static int lu_device_is_mdd(struct lu_device *d)
 {
@@ -82,12 +87,12 @@ static struct mdd_object* mdo2mddo(struct md_object *mdo)
         return container_of(mdo, struct mdd_object, mod_obj);
 }
 
-static inline struct osd_device_operations *mdd_child_ops(struct mdd_device *d)
+static inline struct dt_device_operations *mdd_child_ops(struct mdd_device *d)
 {
-        return d->mdd_child->osd_ops;
+        return d->mdd_child->dd_ops;
 }
 
-struct lu_object *mdd_object_alloc(struct lu_device *d)
+static struct lu_object *mdd_object_alloc(struct lu_device *d)
 {
         struct mdd_object *mdo;
         ENTRY;
@@ -103,14 +108,14 @@ struct lu_object *mdd_object_alloc(struct lu_device *d)
                 return(NULL);
 }
 
-int mdd_object_init(struct lu_object *o)
+static int mdd_object_init(struct lu_object *o)
 {
        struct mdd_device *d = lu2mdd_dev(o->lo_dev);
        struct lu_object  *below;
         struct lu_device  *under;
         ENTRY;
 
-       under = &d->mdd_child->osd_lu_dev;
+       under = &d->mdd_child->dd_lu_dev;
        below = under->ld_ops->ldo_object_alloc(under);
 
         if (below == NULL)
@@ -123,7 +128,7 @@ int mdd_object_init(struct lu_object *o)
         RETURN(0);
 }
 
-void mdd_object_free(struct lu_object *o)
+static void mdd_object_free(struct lu_object *o)
 {
        struct lu_object_header *h;
         struct mdd_object *mdd = mdd_obj(o);
@@ -133,15 +138,112 @@ void mdd_object_free(struct lu_object *o)
         OBD_FREE_PTR(mdd);
 }
 
-void mdd_object_release(struct lu_object *o)
+static int
+mdd_attr_get(struct md_object *obj, void *buf, int buf_len, const char *name,
+             struct context *uctxt)
+{
+        struct mdd_object *mdd_obj = mdo2mddo(obj);
+        struct mdd_device *mdd = mdo2mdd(obj);
+        int rc;
+
+        ENTRY;
+
+        rc = mdd_child_ops(mdd)->dt_attr_get(mdd_object_child(mdd_obj),
+                                             buf, buf_len, name, uctxt);
+        RETURN(rc);
+}
+
+static int
+__mdd_object_destroy(struct mdd_device *mdd, struct mdd_object *obj,
+                     struct thandle *handle)
+{
+        int rc = 0;
+
+        rc = mdd_child_ops(mdd)->dt_object_destroy(mdd_object_child(obj),
+                                                   handle);
+        RETURN(rc);
+}
+
+static int mdd_add_orphan(struct mdd_device *mdd, struct mdd_object *obj,
+                          struct thandle *handle)
+{
+        int rc = 0;
+        ENTRY;
+
+        RETURN(rc);
+}
+
+static int
+open_orphan(struct mdd_object *obj)
+{
+        return 0;
+}
+
+static int
+mdd_add_unlink_log(struct mdd_device *mdd, struct mdd_object *obj,
+                   struct thandle *handle)
+{
+        return 0;
+}
+
+/*
+ * number of blocks to reserve for particular operations. Should be function
+ * of ... something. Stub for now.
+ */
+enum {
+        MDD_OBJECT_DESTROY_CREDITS = 10,
+        MDD_OBJECT_CREATE_CREDITS  = 10,
+        MDD_ATTR_SET_CREDITS       = 10,
+        MDD_INDEX_INSERT_CREDITS   = 10,
+        MDD_INDEX_DELETE_CREDITS   = 10,
+        MDD_LINK_CREDITS           = 10,
+        MDD_RENAME_CREDITS         = 10,
+        MDD_MKDIR_CREDITS          = 10
+};
+
+static int
+mdd_object_destroy(struct md_object *obj)
+{
+        struct mdd_device *mdd = mdo2mdd(obj);
+        struct mdd_object *mdd_obj = mdo2mddo(obj);
+        struct thandle *handle;
+        int rc ;
+        ENTRY;
+
+        handle = mdd_trans_start(mdd, &TXN_PARAM(MDD_OBJECT_DESTROY_CREDITS));
+        if (IS_ERR(handle))
+                RETURN(PTR_ERR(handle));
+
+        mdd_lock(mdd_obj, DT_WRITE_LOCK);
+        if (open_orphan(mdd_obj))
+                rc = mdd_add_orphan(mdd, mdd_obj, handle);
+        else {
+                rc = __mdd_object_destroy(mdd, mdd_obj, handle);
+                if (rc == 0)
+                        rc = mdd_add_unlink_log(mdd, mdd_obj, handle);
+        }
+
+        mdd_unlock(mdd_obj, DT_WRITE_LOCK);
+        mdd_trans_stop(mdd, handle);
+        RETURN(rc);
+}
+
+static void mdd_object_release(struct lu_object *o)
 {
-       struct mdd_device *mdd = lu2mdd_dev(o->lo_dev);
         struct mdd_object *obj = mdd_obj(o);
 
-        mdd_object_put(mdd, obj);
+        int rc;
+        int nlink;
+
+        rc = mdd_attr_get(&obj->mod_obj, &nlink, sizeof(nlink), "NLINK", NULL);
+        if (rc == 0) {
+                if (nlink == 0)
+                        rc = mdd_object_destroy(&obj->mod_obj);
+        } else
+                CERROR("Failed to get nlink: %d. Pretending nonzero.\n", rc);
 }
 
-int mdd_object_print(struct seq_file *f, const struct lu_object *o)
+static int mdd_object_print(struct seq_file *f, const struct lu_object *o)
 {
         return seq_printf(f, LUSTRE_MDD_NAME"-object@%p", o);
 }
@@ -154,44 +256,60 @@ static struct lu_device_operations mdd_lu_ops = {
        .ldo_object_print   = mdd_object_print
 };
 
-struct lu_object* mdd_object_child(struct mdd_object *o)
+static struct dt_object* mdd_object_child(struct mdd_object *o)
+{
+        return container_of(lu_object_next(&o->mod_obj.mo_lu),
+                            struct dt_object, do_lu);
+}
+
+static void mdd_lock(struct mdd_object *obj, enum dt_lock_mode mode)
 {
-       return lu_object_next(&o->mod_obj.mo_lu);
+        struct mdd_device *dev = mdo2mdd(&obj->mod_obj);
+
+        mdd_child_ops(dev)->dt_object_lock(mdd_object_child(obj), mode);
+}
+
+static void mdd_unlock(struct mdd_object *obj, enum dt_lock_mode mode)
+{
+        struct mdd_device *dev = mdo2mdd(&obj->mod_obj);
+
+        mdd_child_ops(dev)->dt_object_unlock(mdd_object_child(obj), mode);
 }
 
-static void
-mdd_lock(struct mdd_device *mdd, struct mdd_object *obj, __u32 mode)
+static void mdd_lock2(struct mdd_object *o0, struct mdd_object *o1)
 {
-        mdd_child_ops(mdd)->osd_object_lock(mdd_object_child(obj), mode);
+        mdd_lock(o0, DT_WRITE_LOCK);
+        mdd_lock(o1, DT_WRITE_LOCK);
 }
 
-static void
-mdd_unlock(struct mdd_device *mdd, struct mdd_object *obj, __u32 mode)
+static void mdd_unlock2(struct mdd_object *o0, struct mdd_object *o1)
 {
-        mdd_child_ops(mdd)->osd_object_unlock(mdd_object_child(obj), mode);
+        mdd_unlock(o0, DT_WRITE_LOCK);
+        mdd_unlock(o1, DT_WRITE_LOCK);
 }
 
-static void* mdd_trans_start(struct mdd_device *mdd, struct mdd_object *obj)
+static struct thandle* mdd_trans_start(struct mdd_device *mdd,
+                                       struct txn_param *p)
 {
-        return mdd_child_ops(mdd)->osd_trans_start(mdd_object_child(obj));
+        return mdd_child_ops(mdd)->dt_trans_start(mdd->mdd_child, p);
 }
 
-static void mdd_trans_stop(struct mdd_device *mdd, void *handle)
+static void mdd_trans_stop(struct mdd_device *mdd, struct thandle *handle)
 {
-        mdd_child_ops(mdd)->osd_trans_stop(handle);
+        mdd_child_ops(mdd)->dt_trans_stop(handle);
 }
 
 static int
 __mdd_object_create(struct mdd_device *mdd, struct mdd_object *pobj,
                     struct mdd_object *child, struct context *uctxt,
-                    void *handle)
+                    struct thandle *handle)
 {
         int rc;
         ENTRY;
 
-        rc = mdd_child_ops(mdd)->osd_object_create(mdd_object_child(pobj),
-                                                   mdd_object_child(child),
-                                                   uctxt, handle);
+        rc = mdd_child_ops(mdd)->dt_object_create(mdd_object_child(pobj),
+                                                  mdd_object_child(child),
+                                                  uctxt, handle);
         /*XXX increase the refcount of the object or not?*/
         RETURN(rc);
 }
@@ -203,13 +321,13 @@ mdd_object_create(struct md_object *pobj, struct md_object *child,
         struct mdd_device *mdd = mdo2mdd(pobj);
         struct mdd_object *mdd_pobj = mdo2mddo(pobj);
         struct mdd_object *mdd_child = mdo2mddo(child);
-        void *handle = NULL;
+        struct thandle *handle;
         int rc;
         ENTRY;
 
-        handle = mdd_trans_start(mdd, mdd_pobj);
-        if (!handle)
-                RETURN(-ENOMEM);
+        handle = mdd_trans_start(mdd, &TXN_PARAM(MDD_OBJECT_CREATE_CREDITS));
+        if (IS_ERR(handle))
+                RETURN(PTR_ERR(handle));
 
         rc = __mdd_object_create(mdd, mdd_pobj, mdd_child, uctxt, handle);
 
@@ -218,79 +336,12 @@ mdd_object_create(struct md_object *pobj, struct md_object *child,
         RETURN(rc);
 }
 
-static int mdd_add_orphan(struct mdd_device *mdd, struct mdd_object *obj,
-                          void *handle)
-{
-        int rc = 0;
-        ENTRY;
-
-        RETURN(rc);
-}
-
-static int
-__mdd_object_destroy(struct mdd_device *mdd, struct mdd_object *obj,
-                    void *handle)
-{
-        int rc = 0;
-
-        rc = mdd_child_ops(mdd)->osd_object_destroy(mdd_object_child(obj),
-                                                    handle);
-        RETURN(rc);
-}
-
-static int
-open_orphan(struct mdd_object *obj)
-{
-        return 0;
-}
-
-static int
-mdd_add_unlink_log(struct mdd_device *mdd, struct mdd_object *obj,
-                   void *handle)
-{
-        return 0;
-}
-
-static int
-mdd_object_destroy(struct md_object *obj)
-{
-        struct mdd_device *mdd = mdo2mdd(obj);
-        struct mdd_object *mdd_obj = mdo2mddo(obj);
-        void *handle;
-        int rc ;
-        ENTRY;
-
-        handle = mdd_trans_start(mdd, mdd_obj);
-        if (!handle)
-                RETURN(-ENOMEM);
-
-        mdd_lock(mdd, mdd_obj, WRITE_LOCK);
-        if (open_orphan(mdd_obj))
-                rc = mdd_add_orphan(mdd, mdd_obj, handle);
-        else {
-                rc = __mdd_object_destroy(mdd, mdd_obj, handle);
-                if (rc)
-                        GOTO(exit, rc);
-
-                rc = mdd_add_unlink_log(mdd, mdd_obj, handle);
-        }
-exit:
-        mdd_unlock(mdd, mdd_obj, WRITE_LOCK);
-        mdd_trans_stop(mdd, handle);
-        RETURN(rc);
-}
-
-void mdd_object_get(struct mdd_device *mdd, struct mdd_object *obj)
-{
-        mdd_child_ops(mdd)->osd_object_get(mdd_object_child(obj));
-}
-
 static int
 __mdd_attr_set(struct mdd_device *mdd, struct mdd_object *obj, void *buf,
                int buf_len, const char *name, struct context *uc_context,
-               void *handle)
+               struct thandle *handle)
 {
-        return mdd_child_ops(mdd)->osd_attr_set(mdd_object_child(obj),
+        return mdd_child_ops(mdd)->dt_attr_set(mdd_object_child(obj),
                                                 buf, buf_len,
                                                 name, uc_context,
                                                 handle);
@@ -301,11 +352,11 @@ mdd_attr_set(struct md_object *obj, void *buf, int buf_len, const char *name,
              struct context *uctxt)
 {
         struct mdd_device *mdd = mdo2mdd(obj);
-        void *handle = NULL;
+        struct thandle *handle;
         int  rc;
         ENTRY;
 
-        handle = mdd_trans_start(mdd, mdo2mddo(obj));
+        handle = mdd_trans_start(mdd, &TXN_PARAM(MDD_ATTR_SET_CREDITS));
         if (!handle)
                 RETURN(-ENOMEM);
 
@@ -317,52 +368,9 @@ mdd_attr_set(struct md_object *obj, void *buf, int buf_len, const char *name,
         RETURN(rc);
 }
 
-static int
-mdd_object_dec_check(struct mdd_device *mdd, struct mdd_object *obj)
-{
-        return mdd_child_ops(mdd)->osd_object_dec_check(mdd_object_child(obj));
-}
-
-
 static struct lu_fid *mdd_object_getfid(struct mdd_object *obj)
 {
-        return &(obj->mod_obj.mo_lu.lo_header->loh_fid);
-}
-
-static int
-mdd_attr_get(struct md_object *obj, void *buf, int buf_len, const char *name,
-             struct context *uctxt)
-{
-        struct mdd_object *mdd_obj = mdo2mddo(obj);
-        struct mdd_device *mdd = mdo2mdd(obj);
-        int rc;
-
-        ENTRY;
-
-        mdd_object_get(mdd, mdd_obj);
-        rc = mdd_child_ops(mdd)->osd_attr_get(mdd_object_child(mdd_obj),
-                                              buf, buf_len, name, uctxt);
-        mdd_object_put(mdd, mdd_obj);
-        RETURN(rc);
-}
-
-int mdd_object_put(struct mdd_device *mdd, struct mdd_object *obj)
-{
-        int rc = 0;
-
-        if ((mdd_object_dec_check(mdd, obj)) == 0) {
-                int nlink;
-
-                rc = mdd_attr_get(&obj->mod_obj, &nlink, sizeof(nlink), "NLINK",
-                                  NULL);
-                if (!rc)
-                        RETURN(-EINVAL);
-
-                if (nlink == 0)
-                        rc = mdd_object_destroy(&obj->mod_obj);
-        }
-
-        RETURN(rc);
+        return lu_object_fid(&obj->mod_obj.mo_lu);
 }
 
 static int
@@ -373,16 +381,12 @@ __mdd_index_insert(struct mdd_device *mdd, struct mdd_object *pobj,
         int rc;
         ENTRY;
 
-        mdd_object_get(mdd, pobj);
-        mdd_lock(mdd, pobj, WRITE_LOCK);
-        mdd_lock(mdd, obj, WRITE_LOCK);
+        mdd_lock2(pobj, obj);
 
-        rc = mdd_child_ops(mdd)->osd_index_insert(mdd_object_child(pobj),
+        rc = mdd_child_ops(mdd)->dt_index_insert(mdd_object_child(pobj),
                                              mdd_object_getfid(obj), name,
                                              uctxt, handle);
-        mdd_unlock(mdd, pobj, WRITE_LOCK);
-        mdd_unlock(mdd, obj, WRITE_LOCK);
-        mdd_object_put(mdd, pobj);
+        mdd_unlock2(pobj, obj);
 
         RETURN(rc);
 }
@@ -393,12 +397,12 @@ mdd_index_insert(struct md_object *pobj, struct md_object *obj, const char *name
 {
         struct mdd_device *mdd = mdo2mdd(pobj);
         int rc;
-        void *handle = NULL;
+        struct thandle *handle;
         ENTRY;
 
-        handle = mdd_trans_start(mdd, mdo2mddo(pobj));
-        if (!handle)
-                RETURN(-ENOMEM);
+        handle = mdd_trans_start(mdd, &TXN_PARAM(MDD_INDEX_INSERT_CREDITS));
+        if (IS_ERR(handle))
+                RETURN(PTR_ERR(handle));
 
         rc = __mdd_index_insert(mdd, mdo2mddo(pobj), mdo2mddo(obj), name, uctxt,
                                 handle);
@@ -410,21 +414,17 @@ mdd_index_insert(struct md_object *pobj, struct md_object *obj, const char *name
 static int
 __mdd_index_delete(struct mdd_device *mdd, struct mdd_object *pobj,
                    struct mdd_object *obj, const char *name,
-                   struct context *uctxt, void *handle)
+                   struct context *uctxt, struct thandle *handle)
 {
         int rc;
         ENTRY;
 
-        mdd_object_get(mdd, pobj);
-        mdd_lock(mdd, pobj, WRITE_LOCK);
-        mdd_lock(mdd, obj, WRITE_LOCK);
+        mdd_lock2(pobj, obj);
 
-        rc = mdd_child_ops(mdd)->osd_index_delete(mdd_object_child(pobj),
+        rc = mdd_child_ops(mdd)->dt_index_delete(mdd_object_child(pobj),
                                               mdd_object_getfid(obj), name,
                                               uctxt, handle);
-        mdd_unlock(mdd, pobj, WRITE_LOCK);
-        mdd_unlock(mdd, obj, WRITE_LOCK);
-        mdd_object_put(mdd, pobj);
+        mdd_unlock2(pobj, obj);
 
         RETURN(rc);
 }
@@ -436,13 +436,13 @@ mdd_index_delete(struct md_object *pobj, struct md_object *obj,
         struct mdd_object *mdd_pobj = mdo2mddo(pobj);
         struct mdd_object *mdd_obj = mdo2mddo(obj);
         struct mdd_device *mdd = mdo2mdd(obj);
-        void *handle = NULL;
+        struct thandle *handle;
         int rc;
         ENTRY;
 
-        handle = mdd_trans_start(mdd, mdo2mddo(obj));
-        if (!handle)
-                RETURN(-ENOMEM);
+        handle = mdd_trans_start(mdd, &TXN_PARAM(MDD_INDEX_DELETE_CREDITS));
+        if (IS_ERR(handle))
+                RETURN(PTR_ERR(handle));
 
         rc = __mdd_index_delete(mdd, mdd_pobj, mdd_obj, name, uctxt, handle);
 
@@ -458,16 +458,15 @@ mdd_link(struct md_object *tgt_obj, struct md_object *src_obj,
         struct mdd_object *mdd_tobj = mdo2mddo(tgt_obj);
         struct mdd_object *mdd_sobj = mdo2mddo(src_obj);
         struct mdd_device *mdd = mdo2mdd(src_obj);
-        void *handle = NULL;
+        struct thandle *handle;
         int rc, nlink;
         ENTRY;
 
-        handle = mdd_trans_start(mdd, mdd_sobj);
-        if (!handle)
-                RETURN(-ENOMEM);
+        handle = mdd_trans_start(mdd, &TXN_PARAM(MDD_LINK_CREDITS));
+        if (IS_ERR(handle))
+                RETURN(PTR_ERR(handle));
 
-        mdd_lock(mdd, mdd_tobj, WRITE_LOCK);
-        mdd_lock(mdd, mdd_sobj, WRITE_LOCK);
+        mdd_lock2(mdd_tobj, mdd_sobj);
 
         rc = __mdd_index_insert(mdd, mdd_tobj, mdd_sobj, name, uctxt, handle);
         if (rc)
@@ -479,8 +478,7 @@ mdd_link(struct md_object *tgt_obj, struct md_object *src_obj,
         rc = __mdd_attr_set(mdd, mdd_sobj, &nlink, sizeof(nlink), "NLINK",
                             uctxt, handle);
 exit:
-        mdd_unlock(mdd, mdd_tobj, WRITE_LOCK);
-        mdd_unlock(mdd, mdd_sobj, WRITE_LOCK);
+        mdd_unlock2(mdd_tobj, mdd_sobj);
 
         mdd_trans_stop(mdd, handle);
         RETURN(rc);
@@ -511,11 +509,11 @@ mdd_rename(struct md_object *src_pobj, struct md_object *tgt_pobj,
         struct mdd_object *mdd_sobj = mdo2mddo(sobj);
         struct mdd_object *mdd_tobj = mdo2mddo(tobj);
         int rc;
-        void *handle = NULL;
+        struct thandle *handle;
 
-        handle = mdd_trans_start(mdd, mdd_spobj);
-        if (!handle)
-                RETURN(-ENOMEM);
+        handle = mdd_trans_start(mdd, &TXN_PARAM(MDD_RENAME_CREDITS));
+        if (IS_ERR(handle))
+                RETURN(PTR_ERR(handle));
 
         mdd_rename_lock(mdd, mdd_spobj, mdd_tpobj, mdd_sobj, mdd_tobj);
 
@@ -544,15 +542,15 @@ static int
 mdd_mkdir(struct md_object *pobj, const char *name, struct md_object *child)
 {
         struct mdd_device *mdd = mdo2mdd(pobj);
-        void *handle;
+        struct thandle *handle;
         int rc = 0;
         ENTRY;
 
-        handle = mdd_trans_start(mdd, mdo2mddo(pobj));
-        if (!handle)
-                RETURN(-ENOMEM);
+        handle = mdd_trans_start(mdd, &TXN_PARAM(MDD_MKDIR_CREDITS));
+        if (IS_ERR(handle))
+                RETURN(PTR_ERR(handle));
 
-        mdd_lock(mdd, mdo2mddo(pobj), WRITE_LOCK);
+        mdd_lock(mdo2mddo(pobj), DT_WRITE_LOCK);
 
         rc = __mdd_object_create(mdd, mdo2mddo(pobj), mdo2mddo(child), NULL,
                                  handle);
@@ -564,7 +562,7 @@ mdd_mkdir(struct md_object *pobj, const char *name, struct md_object *child)
         if (rc)
                 GOTO(cleanup, rc);
 cleanup:
-        mdd_unlock(mdd, mdo2mddo(pobj), WRITE_LOCK);
+        mdd_unlock(mdo2mddo(pobj), DT_WRITE_LOCK);
         mdd_trans_stop(mdd, handle);
         RETURN(rc);
 }
@@ -572,7 +570,7 @@ cleanup:
 static int mdd_root_get(struct md_device *m, struct lu_fid *f)
 {
         struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev);
-        memcpy(f, &mdd->mdd_rootfid, sizeof(*f));
+        *f = mdd->mdd_rootfid;
         return 0;
 }
 

diff --git a/lustre/mdd/mdd_internal.h b/lustre/mdd/mdd_internal.h
index 38203e3..de3d2fa 100644 (file)
--- a/lustre/mdd/mdd_internal.h
+++ b/lustre/mdd/mdd_internal.h
@@ -7,16 +7,9 @@
 #define LUSTRE_MDD_NAME "mdd"
 #define LUSTRE_OSD_NAME "osd"
 
-enum {
-        WRITE_LOCK = 1,
-        READ_LOCK = 2,
-};
-
-
 struct mdd_device {
-        /* NB this field MUST be first */
         struct md_device                 mdd_md_dev;
-        struct osd_device                *mdd_child;
+        struct dt_device                *mdd_child;
         int                              mdd_max_mddize;
         int                              mdd_max_cookiesize;
         struct file                     *mdd_rcvd_filp;