LU-5509 osd: get PFID from linkEA for remote dir on ldiskfs

On the ldiskfs backend, for a directory which parent resides on
remote MDT, to satisfy the local e2fsck, we insert it into the
/REMOTE_PARENT_DIR locally. On the other hand, to make the lookup(..)
on the directory can return the real parent FID, we append the real
parent FID after its ".." name entry in the /REMOTE_PARENT_DIR.

Unfortunately, such PFID-in-dirent cannot be preserved via file-level
backup. So after the restore, we cannot get the right parent FID from
its ".." name entry in the /REMOTE_PARENT_DIR. Under such case, since
we have stored the real parent FID in the directory object's linkEA,
we can parse the linkEA for the real parent FID.

Signed-off-by: Fan Yong <fan.yong@intel.com>
Change-Id: Icf1e24ec911818b3a49a253f67c72334a4b75712
Reviewed-on: http://review.whamcloud.com/11485
Reviewed-by: Andreas Dilger <andreas.dilger@intel.com>
Tested-by: Jenkins
Reviewed-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/obdclass/linkea.c b/lustre/obdclass/linkea.c
index 18e3d16..c3f7e6d 100644 (file)
--- a/lustre/obdclass/linkea.c
+++ b/lustre/obdclass/linkea.c
@@ -97,8 +97,10 @@ void linkea_entry_unpack(const struct link_ea_entry *lee, int *reclen,
        *reclen = (lee->lee_reclen[0] << 8) | lee->lee_reclen[1];
        memcpy(pfid, &lee->lee_parent_fid, sizeof(*pfid));
        fid_be_to_cpu(pfid, pfid);
-       lname->ln_name = lee->lee_name;
-       lname->ln_namelen = *reclen - sizeof(struct link_ea_entry);
+       if (lname != NULL) {
+               lname->ln_name = lee->lee_name;
+               lname->ln_namelen = *reclen - sizeof(struct link_ea_entry);
+       }
 }
 EXPORT_SYMBOL(linkea_entry_unpack);
 

diff --git a/lustre/osd-ldiskfs/osd_handler.c b/lustre/osd-ldiskfs/osd_handler.c
index c85d7d5..03f1d3f 100644 (file)
--- a/lustre/osd-ldiskfs/osd_handler.c
+++ b/lustre/osd-ldiskfs/osd_handler.c
@@ -73,6 +73,7 @@
 #include <lustre_quota.h>
 
 #include <ldiskfs/xattr.h>
+#include <lustre_linkea.h>
 
 int ldiskfs_pdo = 1;
 CFS_MODULE_PARM(ldiskfs_pdo, "i", int, 0644,
@@ -4131,6 +4132,83 @@ int osd_add_oi_cache(struct osd_thread_info *info, struct osd_device *osd,
 }
 
 /**
+ * Get parent FID from the linkEA.
+ *
+ * For a directory which parent resides on remote MDT, to satisfy the
+ * local e2fsck, we insert it into the /REMOTE_PARENT_DIR locally. On
+ * the other hand, to make the lookup(..) on the directory can return
+ * the real parent FID, we append the real parent FID after its ".."
+ * name entry in the /REMOTE_PARENT_DIR.
+ *
+ * Unfortunately, such PFID-in-dirent cannot be preserved via file-level
+ * backup. So after the restore, we cannot get the right parent FID from
+ * its ".." name entry in the /REMOTE_PARENT_DIR. Under such case, since
+ * we have stored the real parent FID in the directory object's linkEA,
+ * we can parse the linkEA for the real parent FID.
+ *
+ * \param[in] env      pointer to the thread context
+ * \param[in] obj      pointer to the object to be handled
+ * \param[out]fid      pointer to the buffer to hold the parent FID
+ *
+ * \retval             0 for getting the real parent FID successfully
+ * \retval             negative error number on failure
+ */
+static int osd_get_pfid_from_linkea(const struct lu_env *env,
+                                   struct osd_object *obj,
+                                   struct lu_fid *fid)
+{
+       struct osd_thread_info  *oti    = osd_oti_get(env);
+       struct lu_buf           *buf    = &oti->oti_big_buf;
+       struct dentry           *dentry = &oti->oti_obj_dentry;
+       struct inode            *inode  = obj->oo_inode;
+       struct linkea_data       ldata  = { 0 };
+       int                      rc;
+       ENTRY;
+
+       fid_zero(fid);
+       if (!S_ISDIR(inode->i_mode))
+               RETURN(-EIO);
+
+again:
+       rc = __osd_xattr_get(inode, dentry, XATTR_NAME_LINK,
+                            buf->lb_buf, buf->lb_len);
+       if (rc == -ERANGE) {
+               rc = __osd_xattr_get(inode, dentry, XATTR_NAME_LINK,
+                                    NULL, 0);
+               if (rc > 0) {
+                       lu_buf_realloc(buf, rc);
+                       if (buf->lb_buf == NULL)
+                               RETURN(-ENOMEM);
+
+                       goto again;
+               }
+       }
+
+       if (unlikely(rc == 0))
+               RETURN(-ENODATA);
+
+       if (rc < 0)
+               RETURN(rc);
+
+       if (unlikely(buf->lb_buf == NULL)) {
+               lu_buf_realloc(buf, rc);
+               if (buf->lb_buf == NULL)
+                       RETURN(-ENOMEM);
+
+               goto again;
+       }
+
+       ldata.ld_buf = buf;
+       rc = linkea_init(&ldata);
+       if (rc == 0) {
+               linkea_first_entry(&ldata);
+               linkea_entry_unpack(ldata.ld_lee, &ldata.ld_reclen, NULL, fid);
+       }
+
+       RETURN(rc);
+}
+
+/**
  * Calls ->lookup() to find dentry. From dentry get inode and
  * read inode's ea to get fid. This is required for  interoperability
  * mode (b11826)
@@ -4184,10 +4262,18 @@ static int osd_ea_lookup_rec(const struct lu_env *env, struct osd_object *obj,
 
                /* done with de, release bh */
                brelse(bh);
-               if (rc != 0)
-                       rc = osd_ea_fid_get(env, obj, ino, fid, id);
-               else
+               if (rc != 0) {
+                       if (unlikely(ino == osd_remote_parent_ino(dev)))
+                               /* If the parent is on remote MDT, and there
+                                * is no FID-in-dirent, then we have to get
+                                * the parent FID from the linkEA.  */
+                               rc = osd_get_pfid_from_linkea(env, obj, fid);
+                       else
+                               rc = osd_ea_fid_get(env, obj, ino, fid, id);
+               } else {
                        osd_id_gen(id, ino, OSD_OII_NOGEN);
+               }
+
                if (rc != 0) {
                        fid_zero(&oic->oic_fid);
                        GOTO(out, rc);
@@ -5225,10 +5311,6 @@ again:
                GOTO(out_journal, rc);
        }
 
-       /* skip the REMOTE_PARENT_DIR. */
-       if (inode == dev->od_mdt_map->omm_remote_parent->d_inode)
-               GOTO(out_inode, rc = 0);
-
        rc = osd_get_lma(info, inode, &info->oti_obj_dentry, lma);
        if (rc == 0) {
                LASSERT(!(lma->lma_compat & LMAC_NOT_IN_OI));
@@ -5408,11 +5490,15 @@ static inline int osd_it_ea_rec(const struct lu_env *env,
        int                     rc    = 0;
        ENTRY;
 
+       LASSERT(obj->oo_inode != dev->od_mdt_map->omm_remote_parent->d_inode);
+
        if (attr & LUDA_VERIFY) {
-               attr |= LUDA_TYPE;
-               if (unlikely(ino == osd_sb(dev)->s_root->d_inode->i_ino)) {
+               if (unlikely(ino == osd_remote_parent_ino(dev))) {
                        attr |= LUDA_IGNORE;
-                       rc = 0;
+                       /* If the parent is on remote MDT, and there
+                        * is no FID-in-dirent, then we have to get
+                        * the parent FID from the linkEA.  */
+                       osd_get_pfid_from_linkea(env, obj, fid);
                } else {
                        rc = osd_dirent_check_repair(env, obj, it, fid, id,
                                                     &attr);
@@ -5426,7 +5512,13 @@ static inline int osd_it_ea_rec(const struct lu_env *env,
                                   it->oie_dirent->oied_name[1] != '.'))
                                RETURN(-ENOENT);
 
-                       rc = osd_ea_fid_get(env, obj, ino, fid, id);
+                       if (unlikely(ino == osd_remote_parent_ino(dev)))
+                               /* If the parent is on remote MDT, and there
+                                * is no FID-in-dirent, then we have to get
+                                * the parent FID from the linkEA.  */
+                               rc = osd_get_pfid_from_linkea(env, obj, fid);
+                       else
+                               rc = osd_ea_fid_get(env, obj, ino, fid, id);
                } else {
                        osd_id_gen(id, ino, OSD_OII_NOGEN);
                }
@@ -5614,6 +5706,7 @@ static void osd_key_fini(const struct lu_context *ctx,
        OBD_FREE(info->oti_it_ea_buf, OSD_IT_EA_BUFSIZE);
        lu_buf_free(&info->oti_iobuf.dr_pg_buf);
        lu_buf_free(&info->oti_iobuf.dr_bl_buf);
+       lu_buf_free(&info->oti_big_buf);
        OBD_FREE_PTR(info);
 }
 

diff --git a/lustre/osd-ldiskfs/osd_internal.h b/lustre/osd-ldiskfs/osd_internal.h
index 06ffc8a..ce78da1 100644 (file)
--- a/lustre/osd-ldiskfs/osd_internal.h
+++ b/lustre/osd-ldiskfs/osd_internal.h
@@ -549,6 +549,7 @@ struct osd_thread_info {
         int                    oti_txns;
         /** used in osd_fid_set() to put xattr */
         struct lu_buf          oti_buf;
+       struct lu_buf          oti_big_buf;
         /** used in osd_ea_fid_set() to set fid into common ea */
        union {
                struct lustre_mdt_attrs oti_mdt_attrs;
@@ -1081,6 +1082,11 @@ static inline int fid_is_internal(const struct lu_fid *fid)
        return (!fid_is_namespace_visible(fid) && !fid_is_idif(fid));
 }
 
+static inline unsigned long osd_remote_parent_ino(struct osd_device *dev)
+{
+       return dev->od_mdt_map->omm_remote_parent->d_inode->i_ino;
+}
+
 #ifdef JOURNAL_START_HAS_3ARGS
 # define osd_journal_start_sb(sb, type, nblock) \
                ldiskfs_journal_start_sb(sb, type, nblock)

diff --git a/lustre/osd-ldiskfs/osd_scrub.c b/lustre/osd-ldiskfs/osd_scrub.c
index edffbe4..518f593 100644 (file)
--- a/lustre/osd-ldiskfs/osd_scrub.c
+++ b/lustre/osd-ldiskfs/osd_scrub.c
@@ -903,6 +903,12 @@ static int osd_iit_iget(struct osd_thread_info *info, struct osd_device *dev,
        int           rc;
        ENTRY;
 
+       /* Not handle the backend root object and agent parent object.
+        * They are neither visible to namespace nor have OI mappings. */
+       if (unlikely(pos == osd_sb(dev)->s_root->d_inode->i_ino ||
+                    pos == osd_remote_parent_ino(dev)))
+               RETURN(SCRUB_NEXT_CONTINUE);
+
        osd_id_gen(lid, pos, OSD_OII_NOGEN);
        inode = osd_iget(info, dev, lid);
        if (IS_ERR(inode)) {