LU-4098 lmv: Match MDT where the FID locates first

With DNE every object can have two locks in different namespaces:
lookup lock in space of MDT storing direntry and update/open lock
in space of MDT storing inode. In lmv_find_cbdata/lmv_lock_lock,
it should try the MDT that the FID maps to first, since this can
be easily found, and only try others if that fails.

In the error handler of lmv_add_targets, it should check whether
ld_tgt_count is being increased before ld_tgt_count is being -1.

Signed-off-by: wang di <di.wang@intel.com>
Change-Id: Idf90ff97421102da97febb0aebaff3f702ff5df7
Reviewed-on: http://review.whamcloud.com/8019
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Andreas Dilger <andreas.dilger@intel.com>
Reviewed-by: Fan Yong <fan.yong@intel.com>

diff --git a/lustre/lmv/lmv_internal.h b/lustre/lmv/lmv_internal.h
index c3485b4..60ed76f 100644 (file)
--- a/lustre/lmv/lmv_internal.h
+++ b/lustre/lmv/lmv_internal.h
@@ -74,35 +74,55 @@ int lmv_revalidate_slaves(struct obd_export *exp, struct mdt_body *mbody,
                          int extra_lock_flags);
 
 static inline struct lmv_tgt_desc *
-lmv_get_target(struct lmv_obd *lmv, mdsno_t mds)
+lmv_get_target(struct lmv_obd *lmv, mdsno_t mdt_idx, int *index)
 {
-       int count = lmv->desc.ld_tgt_count;
        int i;
 
-       for (i = 0; i < count; i++) {
+       for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
                if (lmv->tgts[i] == NULL)
                        continue;
 
-               if (lmv->tgts[i]->ltd_idx == mds)
+               if (lmv->tgts[i]->ltd_idx == mdt_idx) {
+                       if (index != NULL)
+                               *index = i;
                        return lmv->tgts[i];
+               }
        }
 
        return ERR_PTR(-ENODEV);
 }
 
+static inline int
+lmv_find_target_index(struct lmv_obd *lmv, const struct lu_fid *fid)
+{
+       struct lmv_tgt_desc     *ltd;
+       mdsno_t                 mdt_idx = 0;
+       int                     index = 0;
+
+       if (lmv->desc.ld_tgt_count > 1) {
+               int rc;
+               rc = lmv_fld_lookup(lmv, fid, &mdt_idx);
+               if (rc < 0)
+                       return rc;
+       }
+
+       ltd = lmv_get_target(lmv, mdt_idx, &index);
+       if (IS_ERR(ltd))
+               return PTR_ERR(ltd);
+
+       return index;
+}
+
 static inline struct lmv_tgt_desc *
 lmv_find_target(struct lmv_obd *lmv, const struct lu_fid *fid)
 {
-        mdsno_t mds = 0;
-        int rc;
+       int index;
 
-        if (lmv->desc.ld_tgt_count > 1) {
-                rc = lmv_fld_lookup(lmv, fid, &mds);
-                if (rc)
-                        return ERR_PTR(rc);
-        }
+       index = lmv_find_target_index(lmv, fid);
+       if (index < 0)
+               return ERR_PTR(index);
 
-        return lmv_get_target(lmv, mds);
+       return lmv->tgts[index];
 }
 
 static inline int lmv_stripe_md_size(int stripe_count)

diff --git a/lustre/lmv/lmv_obd.c b/lustre/lmv/lmv_obd.c
index c294acd..105d8b2 100644 (file)
--- a/lustre/lmv/lmv_obd.c
+++ b/lustre/lmv/lmv_obd.c
@@ -514,6 +514,7 @@ static int lmv_add_target(struct obd_device *obd, struct obd_uuid *uuidp,
 {
         struct lmv_obd      *lmv = &obd->u.lmv;
         struct lmv_tgt_desc *tgt;
+       int                  orig_tgt_count = 0;
         int                  rc = 0;
         ENTRY;
 
@@ -585,14 +586,17 @@ static int lmv_add_target(struct obd_device *obd, struct obd_uuid *uuidp,
        tgt->ltd_uuid = *uuidp;
        tgt->ltd_active = 0;
        lmv->tgts[index] = tgt;
-       if (index >= lmv->desc.ld_tgt_count)
+       if (index >= lmv->desc.ld_tgt_count) {
+               orig_tgt_count = lmv->desc.ld_tgt_count;
                lmv->desc.ld_tgt_count = index + 1;
+       }
 
        if (lmv->connected) {
                rc = lmv_connect_mdc(obd, tgt);
-               if (rc) {
+               if (rc != 0) {
                        spin_lock(&lmv->lmv_lock);
-                       lmv->desc.ld_tgt_count--;
+                       if (lmv->desc.ld_tgt_count == index + 1)
+                               lmv->desc.ld_tgt_count = orig_tgt_count;
                        memset(tgt, 0, sizeof(*tgt));
                        spin_unlock(&lmv->lmv_lock);
                } else {
@@ -1353,7 +1357,7 @@ int __lmv_fid_alloc(struct lmv_obd *lmv, struct lu_fid *fid,
        int                      rc;
        ENTRY;
 
-       tgt = lmv_get_target(lmv, mds);
+       tgt = lmv_get_target(lmv, mds, NULL);
        if (IS_ERR(tgt))
                RETURN(PTR_ERR(tgt));
 
@@ -1700,27 +1704,39 @@ static int lmv_null_inode(struct obd_export *exp, const struct lu_fid *fid)
 static int lmv_find_cbdata(struct obd_export *exp, const struct lu_fid *fid,
                            ldlm_iterator_t it, void *data)
 {
-        struct obd_device   *obd = exp->exp_obd;
-        struct lmv_obd      *lmv = &obd->u.lmv;
-       __u32                i;
-        int                  rc;
-        ENTRY;
+       struct obd_device       *obd = exp->exp_obd;
+       struct lmv_obd          *lmv = &obd->u.lmv;
+       int                     i;
+       int                     tgt;
+       int                     rc;
+       ENTRY;
 
-        rc = lmv_check_connect(obd);
-        if (rc)
-                RETURN(rc);
+       rc = lmv_check_connect(obd);
+       if (rc)
+               RETURN(rc);
 
-        CDEBUG(D_INODE, "CBDATA for "DFID"\n", PFID(fid));
+       CDEBUG(D_INODE, "CBDATA for "DFID"\n", PFID(fid));
 
        /*
         * With DNE every object can have two locks in different namespaces:
         * lookup lock in space of MDT storing direntry and update/open lock in
-        * space of MDT storing inode.
+        * space of MDT storing inode.  Try the MDT that the FID maps to first,
+        * since this can be easily found, and only try others if that fails.
         */
-       for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
-               if (lmv->tgts[i] == NULL || lmv->tgts[i]->ltd_exp == NULL)
+       for (i = 0, tgt = lmv_find_target_index(lmv, fid);
+            i < lmv->desc.ld_tgt_count;
+            i++, tgt = (tgt + 1) % lmv->desc.ld_tgt_count) {
+               if (tgt < 0) {
+                       CDEBUG(D_HA, "%s: "DFID" is inaccessible: rc = %d\n",
+                              obd->obd_name, PFID(fid), tgt);
+                       tgt = 0;
+               }
+
+               if (lmv->tgts[tgt] == NULL ||
+                   lmv->tgts[tgt]->ltd_exp == NULL)
                        continue;
-               rc = md_find_cbdata(lmv->tgts[i]->ltd_exp, fid, it, data);
+
+               rc = md_find_cbdata(lmv->tgts[tgt]->ltd_exp, fid, it, data);
                if (rc)
                        RETURN(rc);
        }
@@ -1770,7 +1786,7 @@ lmv_locate_target_for_name(struct lmv_obd *lmv, struct lmv_stripe_md *lsm,
                RETURN((void *)oinfo);
        *fid = oinfo->lmo_fid;
        *mds = oinfo->lmo_mds;
-       tgt = lmv_get_target(lmv, *mds);
+       tgt = lmv_get_target(lmv, *mds, NULL);
 
        CDEBUG(D_INFO, "locate on mds %u "DFID"\n", *mds, PFID(fid));
        return tgt;
@@ -2403,7 +2419,7 @@ static int lmv_read_striped_entry(struct obd_export *exp,
        for (i = 0; i < stripe_count; i++) {
                struct page *page = NULL;
 
-               tgt = lmv_get_target(lmv, lsm->lsm_md_oinfo[i].lmo_mds);
+               tgt = lmv_get_target(lmv, lsm->lsm_md_oinfo[i].lmo_mds, NULL);
                if (IS_ERR(tgt))
                        GOTO(out, rc = PTR_ERR(tgt));
 
@@ -3087,33 +3103,42 @@ ldlm_mode_t lmv_lock_match(struct obd_export *exp, __u64 flags,
                            ldlm_policy_data_t *policy, ldlm_mode_t mode,
                            struct lustre_handle *lockh)
 {
-        struct obd_device       *obd = exp->exp_obd;
-        struct lmv_obd          *lmv = &obd->u.lmv;
-        ldlm_mode_t              rc;
-       __u32                    i;
-        ENTRY;
+       struct obd_device       *obd = exp->exp_obd;
+       struct lmv_obd          *lmv = &obd->u.lmv;
+       ldlm_mode_t             rc;
+       int                     tgt;
+       int                     i;
+       ENTRY;
 
-        CDEBUG(D_INODE, "Lock match for "DFID"\n", PFID(fid));
+       CDEBUG(D_INODE, "Lock match for "DFID"\n", PFID(fid));
 
         /*
-         * With CMD every object can have two locks in different namespaces:
-         * lookup lock in space of mds storing direntry and update/open lock in
-         * space of mds storing inode. Thus we check all targets, not only that
-         * one fid was created in.
-         */
-        for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
-               struct lmv_tgt_desc *tgt = lmv->tgts[i];
+        * With DNE every object can have two locks in different namespaces:
+        * lookup lock in space of MDT storing direntry and update/open lock in
+        * space of MDT storing inode.  Try the MDT that the FID maps to first,
+        * since this can be easily found, and only try others if that fails.
+        */
+       for (i = 0, tgt = lmv_find_target_index(lmv, fid);
+            i < lmv->desc.ld_tgt_count;
+            i++, tgt = (tgt + 1) % lmv->desc.ld_tgt_count) {
+               if (tgt < 0) {
+                       CDEBUG(D_HA, "%s: "DFID" is inaccessible: rc = %d\n",
+                              obd->obd_name, PFID(fid), tgt);
+                       tgt = 0;
+               }
 
-               if (tgt == NULL || tgt->ltd_exp == NULL || !tgt->ltd_active)
+               if (lmv->tgts[tgt] == NULL ||
+                   lmv->tgts[tgt]->ltd_exp == NULL ||
+                   lmv->tgts[tgt]->ltd_active == 0)
                        continue;
 
-               rc = md_lock_match(tgt->ltd_exp, flags, fid, type, policy, mode,
-                                  lockh);
-                if (rc)
-                        RETURN(rc);
-        }
+               rc = md_lock_match(lmv->tgts[tgt]->ltd_exp, flags, fid,
+                                  type, policy, mode, lockh);
+               if (rc)
+                       RETURN(rc);
+       }
 
-        RETURN(0);
+       RETURN(0);
 }
 
 int lmv_get_lustre_md(struct obd_export *exp, struct ptlrpc_request *req,