Whamcloud - gitweb
LU-80 lov: large stripe count support
[fs/lustre-release.git] / lustre / mdd / mdd_lov.c
index 6044429..69df68e 100644 (file)
  * GPL HEADER END
  */
 /*
- * Copyright  2008 Sun Microsystems, Inc. All rights reserved
+ * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
+ *
+ * Copyright (c) 2011 Whamcloud, Inc.
+ *
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -60,7 +63,7 @@
 #include "mdd_internal.h"
 
 static int mdd_notify(struct obd_device *host, struct obd_device *watched,
-                      enum obd_notify_event ev, void *owner)
+                      enum obd_notify_event ev, void *owner, void *data)
 {
         struct mdd_device *mdd = owner;
         int rc = 0;
@@ -72,14 +75,17 @@ static int mdd_notify(struct obd_device *host, struct obd_device *watched,
                 case OBD_NOTIFY_ACTIVE:
                 case OBD_NOTIFY_SYNC:
                 case OBD_NOTIFY_SYNC_NONBLOCK:
-                        rc = md_do_upcall(NULL, &mdd->mdd_md_dev, MD_LOV_SYNC);
+                        rc = md_do_upcall(NULL, &mdd->mdd_md_dev,
+                                          MD_LOV_SYNC, data);
                         break;
                 case OBD_NOTIFY_CONFIG:
-                        rc = md_do_upcall(NULL, &mdd->mdd_md_dev, MD_LOV_CONFIG);
+                        rc = md_do_upcall(NULL, &mdd->mdd_md_dev,
+                                          MD_LOV_CONFIG, data);
                         break;
 #ifdef HAVE_QUOTA_SUPPORT
                 case OBD_NOTIFY_QUOTA:
-                        rc = md_do_upcall(NULL, &mdd->mdd_md_dev, MD_LOV_QUOTA);
+                        rc = md_do_upcall(NULL, &mdd->mdd_md_dev,
+                                          MD_LOV_QUOTA, data);
                         break;
 #endif
                 default:
@@ -115,11 +121,11 @@ int mdd_init_obd(const struct lu_env *env, struct mdd_device *mdd,
         if (!bufs)
                 GOTO(cleanup_mem, rc = -ENOMEM);
 
-        snprintf(name, strlen(MDD_OBD_NAME) + 35, "%s-%s-%d",
-                 MDD_OBD_NAME, dev, mds_id);
+        snprintf(name, strlen(MDD_OBD_NAME) + 35, "%s-%s",
+                 MDD_OBD_NAME, dev);
 
-        snprintf(uuid, strlen(MDD_OBD_UUID) + 35, "%s-%s-%d",
-                 MDD_OBD_UUID, dev, mds_id);
+        snprintf(uuid, strlen(MDD_OBD_UUID) + 35, "%s-%s",
+                 MDD_OBD_UUID, dev);
 
         lustre_cfg_bufs_reset(bufs, name);
         lustre_cfg_bufs_set_string(bufs, 1, MDD_OBD_TYPE);
@@ -142,8 +148,13 @@ int mdd_init_obd(const struct lu_env *env, struct mdd_device *mdd,
                 LBUG();
         }
 
+        cfs_spin_lock(&obd->obd_dev_lock);
         obd->obd_recovering = 1;
+        cfs_spin_unlock(&obd->obd_dev_lock);
         obd->u.mds.mds_id = mds_id;
+        obd->u.obt.obt_osd_properties.osd_max_ea_size =
+                                               mdd->mdd_dt_conf.ddp_max_ea_size;
+
         rc = class_setup(obd, lcfg);
         if (rc)
                 GOTO(class_detach, rc);
@@ -155,6 +166,7 @@ int mdd_init_obd(const struct lu_env *env, struct mdd_device *mdd,
         obd->obd_upcall.onu_upcall = mdd_notify;
         obd->obd_upcall.onu_owner = mdd;
         mdd->mdd_obd_dev = obd;
+
         EXIT;
 class_detach:
         if (rc)
@@ -236,7 +248,7 @@ static int mdd_lov_set_stripe_md(const struct lu_env *env,
 {
         struct mdd_device       *mdd = mdo2mdd(&obj->mod_obj);
         struct obd_device       *obd = mdd2obd_dev(mdd);
-        struct obd_export       *lov_exp = obd->u.mds.mds_osc_exp;
+        struct obd_export       *lov_exp = obd->u.mds.mds_lov_exp;
         struct lov_stripe_md    *lsm = NULL;
         int rc;
         ENTRY;
@@ -271,8 +283,8 @@ static int mdd_lov_set_dir_md(const struct lu_env *env,
 
         /* if { size, offset, count } = { 0, -1, 0 } and no pool (i.e. all default
          * values specified) then delete default striping from dir. */
-        if (lum->lmm_stripe_size == 0 && lum->lmm_stripe_count == 0 &&
-            lum->lmm_stripe_offset == (typeof(lum->lmm_stripe_offset))(-1) &&
+        if (LOVEA_DELETE_VALUES(lum->lmm_stripe_size, lum->lmm_stripe_count,
+                                lum->lmm_stripe_offset) &&
             lum->lmm_magic != LOV_USER_MAGIC_V3) {
                 rc = mdd_xattr_set_txn(env, obj, &LU_BUF_NULL,
                                        XATTR_NAME_LOV, 0, handle);
@@ -310,7 +322,7 @@ int mdd_lov_set_md(const struct lu_env *env, struct mdd_object *pobj,
                    int lmm_size, struct thandle *handle, int set_stripe)
 {
         struct lu_buf *buf;
-        umode_t mode;
+        cfs_umode_t mode;
         int rc = 0;
         ENTRY;
 
@@ -351,16 +363,6 @@ int mdd_lov_set_md(const struct lu_env *env, struct mdd_object *pobj,
         RETURN(rc);
 }
 
-/*
- * XXX: this is for create lsm object id, which should identify the lsm object
- * unique in the whole mds, as I see. But it seems, we still not need it
- * now. Right? So just borrow the ll_fid_build_ino().
- */
-static obd_id mdd_lov_create_id(const struct lu_fid *fid)
-{
-        return fid_flatten(fid);
-}
-
 int mdd_lov_objid_prepare(struct mdd_device *mdd, struct lov_mds_md *lmm)
 {
         /* copy mds_lov code is using wrong layer */
@@ -378,7 +380,7 @@ void mdd_lov_create_finish(const struct lu_env *env, struct mdd_device *mdd,
                            const struct md_op_spec *spec)
 {
         if (lmm && !spec->no_create)
-                OBD_FREE(lmm, lmm_size);
+                OBD_FREE_LARGE(lmm, lmm_size);
 }
 
 int mdd_lov_create(const struct lu_env *env, struct mdd_device *mdd,
@@ -387,12 +389,12 @@ int mdd_lov_create(const struct lu_env *env, struct mdd_device *mdd,
                    const struct md_op_spec *spec, struct lu_attr *la)
 {
         struct obd_device     *obd = mdd2obd_dev(mdd);
-        struct obd_export     *lov_exp = obd->u.mds.mds_osc_exp;
+        struct obd_export     *lov_exp = obd->u.mds.mds_lov_exp;
         struct lu_site        *site = mdd2lu_dev(mdd)->ld_site;
         struct obdo           *oa;
         struct lov_stripe_md  *lsm = NULL;
         const void            *eadata = spec->u.sp_ea.eadata;
-        __u32                  create_flags = spec->sp_cr_flags;
+        __u64                  create_flags = spec->sp_cr_flags;
         struct obd_trans_info *oti = &mdd_env_info(env)->mti_oti;
         int                    rc = 0;
         ENTRY;
@@ -407,7 +409,13 @@ int mdd_lov_create(const struct lu_env *env, struct mdd_device *mdd,
         if (spec->no_create != 0) {
                 *lmm = (struct lov_mds_md *)spec->u.sp_ea.eadata;
                 *lmm_size = spec->u.sp_ea.eadatalen;
-                RETURN(0);
+                if (*lmm_size == lov_mds_md_size((*lmm)->lmm_stripe_count,
+                                                 (*lmm)->lmm_magic)) {
+                        RETURN(0);
+                } else {
+                        CERROR("incorrect lsm received during recovery\n");
+                        RETURN(-EPROTO);
+                }
         }
 
         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ALLOC_OBDO))
@@ -418,9 +426,9 @@ int mdd_lov_create(const struct lu_env *env, struct mdd_device *mdd,
 
         oa->o_uid = 0; /* must have 0 uid / gid on OST */
         oa->o_gid = 0;
-        oa->o_gr = mdt_to_obd_objgrp(lu_site2md(site)->ms_node_id);
+        oa->o_seq = mdt_to_obd_objseq(lu_site2md(site)->ms_node_id);
         oa->o_mode = S_IFREG | 0600;
-        oa->o_id = mdd_lov_create_id(mdd_object_fid(child));
+        oa->o_id = fid_ver_oid(mdd_object_fid(child));
         oa->o_valid = OBD_MD_FLID | OBD_MD_FLTYPE | OBD_MD_FLFLAGS |
                 OBD_MD_FLMODE | OBD_MD_FLUID | OBD_MD_FLGID | OBD_MD_FLGROUP;
         oa->o_size = 0;
@@ -432,8 +440,6 @@ int mdd_lov_create(const struct lu_env *env, struct mdd_device *mdd,
                                            0, &lsm, (void*)eadata);
                         if (rc)
                                 GOTO(out_oti, rc);
-                        lsm->lsm_object_id = oa->o_id;
-                        lsm->lsm_object_gr = oa->o_gr;
                 } else if (parent != NULL) {
                         /* get lov ea from parent and set to lov */
                         struct lov_mds_md *_lmm;
@@ -457,6 +463,7 @@ int mdd_lov_create(const struct lu_env *env, struct mdd_device *mdd,
                                 GOTO(out_oti, rc);
                 }
 
+                OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_OPEN_WAIT_CREATE, 10);
                 rc = obd_create(lov_exp, oa, &lsm, oti);
                 if (rc) {
                         if (rc > 0) {
@@ -466,17 +473,18 @@ int mdd_lov_create(const struct lu_env *env, struct mdd_device *mdd,
                         }
                         GOTO(out_oti, rc);
                 }
-                LASSERT_MDS_GROUP(lsm->lsm_object_gr);
+                LASSERT_SEQ_IS_MDT(lsm->lsm_object_seq);
         } else {
                 LASSERT(eadata != NULL);
                 rc = obd_iocontrol(OBD_IOC_LOV_SETEA, lov_exp, 0, &lsm,
                                    (void*)eadata);
                 if (rc)
                         GOTO(out_oti, rc);
-                lsm->lsm_object_id = oa->o_id;
-                lsm->lsm_object_gr = oa->o_gr;
+
         }
 
+        lsm->lsm_object_id = fid_ver_oid(mdd_object_fid(child));
+        lsm->lsm_object_seq = fid_seq(mdd_object_fid(child));
         /*
          * Sometimes, we may truncate some object(without lsm) then open it
          * (with write flags), so creating lsm above.  The Nonzero(truncated)
@@ -497,9 +505,8 @@ int mdd_lov_create(const struct lu_env *env, struct mdd_device *mdd,
                  * filter_fid, but can not see what is the usages. So just pack
                  * o_seq o_ver here, maybe fix it after this cycle.
                  */
-                oa->o_fid = fid_seq(mdd_object_fid(child));
-                oa->o_generation = fid_oid(mdd_object_fid(child));
-                oa->o_valid |= OBD_MD_FLFID | OBD_MD_FLGENER;
+                obdo_from_inode(oa, NULL,
+                                (struct lu_fid *)mdd_object_fid(child), 0);
                 oinfo->oi_oa = oa;
                 oinfo->oi_md = lsm;
                 oinfo->oi_capa = NULL;
@@ -518,7 +525,6 @@ int mdd_lov_create(const struct lu_env *env, struct mdd_device *mdd,
                         GOTO(out_oti, rc);
                 }
         }
-
         /* blksize should be changed after create data object */
         la->la_valid |= LA_BLKSIZE;
         la->la_blksize = oa->o_blksize;
@@ -558,7 +564,7 @@ int mdd_lovobj_unlink(const struct lu_env *env, struct mdd_device *mdd,
                       int log_unlink)
 {
         struct obd_device     *obd = mdd2obd_dev(mdd);
-        struct obd_export     *lov_exp = obd->u.mds.mds_osc_exp;
+        struct obd_export     *lov_exp = obd->u.mds.mds_lov_exp;
         struct lov_stripe_md  *lsm = NULL;
         struct obd_trans_info *oti = &mdd_env_info(env)->mti_oti;
         struct obdo           *oa = &mdd_env_info(env)->mti_oa;
@@ -579,7 +585,7 @@ int mdd_lovobj_unlink(const struct lu_env *env, struct mdd_device *mdd,
         }
 
         oa->o_id = lsm->lsm_object_id;
-        oa->o_gr = mdt_to_obd_objgrp(lu_site2md(site)->ms_node_id);
+        oa->o_seq = mdt_to_obd_objseq(lu_site2md(site)->ms_node_id);
         oa->o_mode = la->la_mode & S_IFMT;
         oa->o_valid = OBD_MD_FLID | OBD_MD_FLTYPE | OBD_MD_FLGROUP;
 
@@ -589,8 +595,8 @@ int mdd_lovobj_unlink(const struct lu_env *env, struct mdd_device *mdd,
                 oti->oti_logcookies = logcookies;
         }
 
-        CDEBUG(D_INFO, "destroying OSS object %d/%d\n",
-                        (int)oa->o_id, (int)oa->o_gr);
+        CDEBUG(D_INFO, "destroying OSS object "LPU64":"LPU64"\n", oa->o_seq,
+               oa->o_id);
 
         rc = obd_destroy(lov_exp, oa, lsm, oti, NULL, NULL);
 
@@ -599,9 +605,8 @@ int mdd_lovobj_unlink(const struct lu_env *env, struct mdd_device *mdd,
 }
 
 /*
- * called with obj not locked. 
+ * called with obj locked. 
  */
-
 int mdd_lov_destroy(const struct lu_env *env, struct mdd_device *mdd,
                     struct mdd_object *obj, struct lu_attr *la)
 {
@@ -609,6 +614,11 @@ int mdd_lov_destroy(const struct lu_env *env, struct mdd_device *mdd,
         int                rc;
         ENTRY;
 
+        LASSERT(mdd_write_locked(env, obj) != 0);
+
+        if (unlikely(!S_ISREG(mdd_object_type(obj))))
+                RETURN(0);
+
         if (unlikely(la->la_nlink != 0)) {
                 CWARN("Attempt to destroy OSS object when nlink == %d\n",
                       la->la_nlink);
@@ -624,8 +634,8 @@ int mdd_lov_destroy(const struct lu_env *env, struct mdd_device *mdd,
 
         /* get lov ea */
 
-        rc = mdd_get_md_locked(env, obj, ma->ma_lmm, &ma->ma_lmm_size,
-                               XATTR_NAME_LOV);
+        rc = mdd_get_md(env, obj, ma->ma_lmm, &ma->ma_lmm_size,
+                        XATTR_NAME_LOV);
 
         if (rc <= 0) {
                 CWARN("Get lov ea failed for "DFID" rc = %d\n",
@@ -636,7 +646,7 @@ int mdd_lov_destroy(const struct lu_env *env, struct mdd_device *mdd,
         }
 
         ma->ma_valid = MA_LOV;
-        
+
         rc = mdd_unlink_log(env, mdd, obj, ma);
         if (rc) {
                 CWARN("mds unlink log for "DFID" failed: %d\n",
@@ -651,54 +661,13 @@ int mdd_lov_destroy(const struct lu_env *env, struct mdd_device *mdd,
         RETURN(rc);
 }
 
-int mdd_log_op_unlink(struct obd_device *obd,
-                      struct lov_mds_md *lmm, int lmm_size,
-                      struct llog_cookie *logcookies, int cookies_size)
-{
-        struct mds_obd *mds = &obd->u.mds;
-        struct lov_stripe_md *lsm = NULL;
-        struct llog_unlink_rec *lur;
-        struct llog_ctxt *ctxt;
-        int rc;
-        ENTRY;
-
-        if (IS_ERR(mds->mds_osc_obd))
-                RETURN(PTR_ERR(mds->mds_osc_obd));
-
-        rc = obd_unpackmd(mds->mds_osc_exp, &lsm, lmm, lmm_size);
-        if (rc < 0)
-                RETURN(rc);
-        rc = obd_checkmd(mds->mds_osc_exp, obd->obd_self_export, lsm);
-        if (rc)
-                GOTO(out, rc);
-        /* first prepare unlink log record */
-        OBD_ALLOC(lur, sizeof(*lur));
-        if (!lur)
-                GOTO(out, rc = -ENOMEM);
-        lur->lur_hdr.lrh_len = lur->lur_tail.lrt_len = sizeof(*lur);
-        lur->lur_hdr.lrh_type = MDS_UNLINK_REC;
-
-        ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
-        rc = llog_add(ctxt, &lur->lur_hdr, lsm, logcookies,
-                      cookies_size / sizeof(struct llog_cookie));
-        llog_ctxt_put(ctxt);
-
-        OBD_FREE(lur, sizeof(*lur));
-        GOTO(out, rc);
-out:
-        obd_free_memmd(mds->mds_osc_exp, &lsm);
-        return rc;
-}
-
 int mdd_unlink_log(const struct lu_env *env, struct mdd_device *mdd,
                    struct mdd_object *mdd_cobj, struct md_attr *ma)
 {
-        struct obd_device *obd = mdd2obd_dev(mdd);
-
         LASSERT(ma->ma_valid & MA_LOV);
 
         if ((ma->ma_cookie_size > 0) &&
-            (mdd_log_op_unlink(obd, ma->ma_lmm, ma->ma_lmm_size,
+            (mds_log_op_unlink(mdd2obd_dev(mdd), ma->ma_lmm, ma->ma_lmm_size,
                                ma->ma_cookie, ma->ma_cookie_size) > 0)) {
                 CDEBUG(D_HA, "DEBUG: unlink log is added for object "DFID"\n",
                        PFID(mdd_object_fid(mdd_cobj)));
@@ -708,8 +677,8 @@ int mdd_unlink_log(const struct lu_env *env, struct mdd_device *mdd,
 }
 
 int mdd_log_op_setattr(struct obd_device *obd, __u32 uid, __u32 gid,
-                      struct lov_mds_md *lmm, int lmm_size,
-                      struct llog_cookie *logcookies, int cookies_size)
+                       struct lov_mds_md *lmm, int lmm_size,
+                       struct llog_cookie *logcookies, int cookies_size)
 {
         struct mds_obd *mds = &obd->u.mds;
         struct lov_stripe_md *lsm = NULL;
@@ -718,17 +687,13 @@ int mdd_log_op_setattr(struct obd_device *obd, __u32 uid, __u32 gid,
         int rc;
         ENTRY;
 
-        if (IS_ERR(mds->mds_osc_obd))
-                RETURN(PTR_ERR(mds->mds_osc_obd));
+        if (IS_ERR(mds->mds_lov_obd))
+                RETURN(PTR_ERR(mds->mds_lov_obd));
 
-        rc = obd_unpackmd(mds->mds_osc_exp, &lsm, lmm, lmm_size);
+        rc = obd_unpackmd(mds->mds_lov_exp, &lsm, lmm, lmm_size);
         if (rc < 0)
                 RETURN(rc);
 
-        rc = obd_checkmd(mds->mds_osc_exp, obd->obd_self_export, lsm);
-        if (rc)
-                GOTO(out, rc);
-
         OBD_ALLOC(lsr, sizeof(*lsr));
         if (!lsr)
                 GOTO(out, rc = -ENOMEM);
@@ -748,7 +713,7 @@ int mdd_log_op_setattr(struct obd_device *obd, __u32 uid, __u32 gid,
 
         OBD_FREE(lsr, sizeof(*lsr));
  out:
-        obd_free_memmd(mds->mds_osc_exp, &lsm);
+        obd_free_memmd(mds->mds_lov_exp, &lsm);
         RETURN(rc);
 }
 
@@ -774,7 +739,7 @@ int mdd_setattr_log(const struct lu_env *env, struct mdd_device *mdd,
 
 static int mdd_osc_setattr_async(struct obd_device *obd, __u32 uid, __u32 gid,
                           struct lov_mds_md *lmm, int lmm_size,
-                          struct llog_cookie *logcookies, __u64 id, __u32 gen,
+                          struct llog_cookie *logcookies, const struct lu_fid *parent,
                           struct obd_capa *oc)
 {
         struct mds_obd *mds = &obd->u.mds;
@@ -793,15 +758,10 @@ static int mdd_osc_setattr_async(struct obd_device *obd, __u32 uid, __u32 gid,
 
         LASSERT(lmm);
 
-        rc = obd_unpackmd(mds->mds_osc_exp, &oinfo.oi_md, lmm, lmm_size);
+        rc = obd_unpackmd(mds->mds_lov_exp, &oinfo.oi_md, lmm, lmm_size);
         if (rc < 0) {
-                CERROR("Error unpack md %p for inode "LPU64"\n", lmm, id);
-                GOTO(out, rc);
-        }
-
-        rc = obd_checkmd(mds->mds_osc_exp, obd->obd_self_export, oinfo.oi_md);
-        if (rc) {
-                CERROR("Error revalidate lsm %p \n", oinfo.oi_md);
+                CERROR("Error unpack md %p for obj "DFID"\n", lmm,
+                        PFID(parent));
                 GOTO(out, rc);
         }
 
@@ -809,7 +769,7 @@ static int mdd_osc_setattr_async(struct obd_device *obd, __u32 uid, __u32 gid,
         oinfo.oi_oa->o_uid = uid;
         oinfo.oi_oa->o_gid = gid;
         oinfo.oi_oa->o_id = oinfo.oi_md->lsm_object_id;
-        oinfo.oi_oa->o_gr = oinfo.oi_md->lsm_object_gr;
+        oinfo.oi_oa->o_seq = oinfo.oi_md->lsm_object_seq;
         oinfo.oi_oa->o_valid |= OBD_MD_FLID | OBD_MD_FLGROUP |
                                 OBD_MD_FLUID | OBD_MD_FLGID;
         if (logcookies) {
@@ -817,19 +777,17 @@ static int mdd_osc_setattr_async(struct obd_device *obd, __u32 uid, __u32 gid,
                 oti.oti_logcookies = logcookies;
         }
 
-        oinfo.oi_oa->o_fid = id;
-        oinfo.oi_oa->o_generation = gen;
-        oinfo.oi_oa->o_valid |= OBD_MD_FLFID | OBD_MD_FLGENER;
+        obdo_from_inode(oinfo.oi_oa, NULL, (struct lu_fid *)parent, 0);
         oinfo.oi_capa = oc;
 
         /* do async setattr from mds to ost not waiting for responses. */
-        rc = obd_setattr_async(mds->mds_osc_exp, &oinfo, &oti, NULL);
+        rc = obd_setattr_async(mds->mds_lov_exp, &oinfo, &oti, NULL);
         if (rc)
                 CDEBUG(D_INODE, "mds to ost setattr objid 0x"LPX64
                        " on ost error %d\n", oinfo.oi_md->lsm_object_id, rc);
 out:
         if (oinfo.oi_md)
-                obd_free_memmd(mds->mds_osc_exp, &oinfo.oi_md);
+                obd_free_memmd(mds->mds_lov_exp, &oinfo.oi_md);
         OBDO_FREE(oinfo.oi_oa);
         RETURN(rc);
 }
@@ -852,7 +810,170 @@ int mdd_lov_setattr_async(const struct lu_env *env, struct mdd_object *obj,
                 RETURN(rc);
 
         rc = mdd_osc_setattr_async(obd, tmp_la->la_uid, tmp_la->la_gid, lmm,
-                                   lmm_size, logcookies, fid_seq(fid),
-                                   fid_oid(fid), NULL);
+                                   lmm_size, logcookies, fid, NULL);
+        RETURN(rc);
+}
+
+static int grouplock_blocking_ast(struct ldlm_lock *lock,
+                                  struct ldlm_lock_desc *desc,
+                                  void *data, int flag)
+{
+        struct md_attr *ma = data;
+        struct lustre_handle lockh;
+        int rc = 0;
+        ENTRY;
+
+        switch (flag)
+        {
+                case LDLM_CB_BLOCKING :
+                        /* lock is canceled */
+                        CDEBUG(D_DLMTRACE, "Lock %p is canceled\n", lock);
+
+                        ldlm_lock2handle(lock, &lockh);
+                        rc = ldlm_cli_cancel(&lockh);
+
+                        break;
+                case LDLM_CB_CANCELING :
+                        CDEBUG(D_DLMTRACE,
+                               "Lock %p has been canceled, do cleaning\n",
+                               lock);
+
+                        if (ma && ma->ma_som)
+                                OBD_FREE_PTR(ma->ma_som);
+                        if (ma)
+                                OBD_FREE_PTR(ma);
+                        break;
+                default:
+                        LBUG();
+        }
+        RETURN(rc);
+}
+
+static int grouplock_glimpse_ast(struct ldlm_lock *lock, void *data)
+{
+        struct ptlrpc_request *req = data;
+        struct ost_lvb *lvb;
+        int rc;
+        struct md_attr *ma;
+        ENTRY;
+
+        ma = lock->l_ast_data;
+
+        req_capsule_extend(&req->rq_pill, &RQF_LDLM_GL_CALLBACK);
+        req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
+                             sizeof(*lvb));
+        rc = req_capsule_server_pack(&req->rq_pill);
+        if (rc) {
+                CERROR("failed pack reply: %d\n", rc);
+                GOTO(out, rc);
+        }
+
+        lvb = req_capsule_server_get(&req->rq_pill, &RMF_DLM_LVB);
+
+        if ((ma) && (ma->ma_valid & MA_SOM)) {
+                lvb->lvb_size = ma->ma_som->msd_size;
+                lvb->lvb_blocks = ma->ma_som->msd_blocks;
+        } else if ((ma) && (ma->ma_valid & MA_INODE)) {
+                lvb->lvb_size = ma->ma_attr.la_size;
+                lvb->lvb_blocks = ma->ma_attr.la_blocks;
+        } else {
+                lvb->lvb_size = 0;
+                rc = -ELDLM_NO_LOCK_DATA;
+        }
+
+        EXIT;
+out:
+        if (rc == -ELDLM_NO_LOCK_DATA)
+                lustre_pack_reply(req, 1, NULL, NULL);
+
+        req->rq_status = rc;
+        return rc;
+}
+
+int mdd_file_lock(const struct lu_env *env, struct md_object *obj,
+                  struct lov_mds_md *lmm, struct ldlm_extent *extent,
+                  struct lustre_handle *lockh)
+{
+        struct ldlm_enqueue_info einfo = { 0 };
+        struct obd_info oinfo = { { { 0 } } };
+        struct obd_device *obd;
+        struct obd_export *lov_exp;
+        struct lov_stripe_md *lsm = NULL;
+        struct md_attr *ma = NULL;
+        int rc;
+        ENTRY;
+
+        obd = mdo2mdd(obj)->mdd_obd_dev;
+        lov_exp = obd->u.mds.mds_lov_exp;
+
+        obd_unpackmd(lov_exp, &lsm, lmm,
+                     lov_mds_md_size(lmm->lmm_stripe_count, lmm->lmm_magic));
+
+        OBD_ALLOC_PTR(ma);
+        if (ma == NULL)
+                GOTO(out, rc = -ENOMEM);
+
+        OBD_ALLOC_PTR(ma->ma_som);
+        if (ma->ma_som == NULL)
+                GOTO(out, rc = -ENOMEM);
+
+        ma->ma_need = MA_SOM | MA_INODE;
+        mo_attr_get(env, obj, ma);
+
+        einfo.ei_type = LDLM_EXTENT;
+        einfo.ei_mode = LCK_GROUP;
+        einfo.ei_cb_bl = grouplock_blocking_ast;
+        einfo.ei_cb_cp = ldlm_completion_ast;
+        einfo.ei_cb_gl = grouplock_glimpse_ast;
+
+        if (ma->ma_valid & (MA_SOM | MA_INODE))
+                einfo.ei_cbdata = ma;
+        else
+                einfo.ei_cbdata = NULL;
+
+        memset(&oinfo.oi_policy, 0, sizeof(oinfo.oi_policy));
+        oinfo.oi_policy.l_extent = *extent;
+        oinfo.oi_lockh = lockh;
+        oinfo.oi_md = lsm;
+        oinfo.oi_flags = 0;
+
+        rc = obd_enqueue(lov_exp, &oinfo, &einfo, NULL);
+        /* ei_cbdata is used as a free flag at exit */
+        if (rc)
+                einfo.ei_cbdata = NULL;
+
+        obd_unpackmd(lov_exp, &lsm, NULL, 0);
+
+out:
+        /* ma is freed if not used as callback data */
+        if ((einfo.ei_cbdata == NULL) && ma && ma->ma_som)
+                OBD_FREE_PTR(ma->ma_som);
+        if ((einfo.ei_cbdata == NULL) && ma)
+                OBD_FREE_PTR(ma);
+
+        RETURN(rc);
+}
+
+int mdd_file_unlock(const struct lu_env *env, struct md_object *obj,
+                    struct lov_mds_md *lmm, struct lustre_handle *lockh)
+{
+        struct obd_device *obd;
+        struct obd_export *lov_exp;
+        struct lov_stripe_md *lsm = NULL;
+        int rc;
+        ENTRY;
+
+        LASSERT(lustre_handle_is_used(lockh));
+
+        obd = mdo2mdd(obj)->mdd_obd_dev;
+        lov_exp = obd->u.mds.mds_lov_exp;
+
+        obd_unpackmd(lov_exp, &lsm, lmm,
+                     lov_mds_md_size(lmm->lmm_stripe_count, lmm->lmm_magic));
+
+        rc = obd_cancel(lov_exp, lsm, LCK_GROUP, lockh);
+
+        obd_unpackmd(lov_exp, &lsm, NULL, 0);
+
         RETURN(rc);
 }