LU-1187 osd: add remote entry insert for ZFS DNE.

Add remote entry insert in osd_index_insert for DNE on ZFS.

Signed-off-by: Wang Di <di.wang@intel.com>
Change-Id: I4c2337e69e4c4c4754b8ff7995dd69318f3d674d
Reviewed-on: http://review.whamcloud.com/4933
Tested-by: Hudson
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Andreas Dilger <andreas.dilger@intel.com>
Reviewed-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>

diff --git a/lustre/osd-zfs/osd_index.c b/lustre/osd-zfs/osd_index.c
index f14faf7..f88958c 100644 (file)
--- a/lustre/osd-zfs/osd_index.c
+++ b/lustre/osd-zfs/osd_index.c
@@ -476,6 +476,27 @@ static inline void osd_object_put(const struct lu_env *env,
        lu_object_put(env, &obj->oo_dt.do_lu);
 }
 
+static int osd_remote_fid(const struct lu_env *env, struct osd_device *osd,
+                         struct lu_fid *fid)
+{
+       struct lu_seq_range     *range = &osd_oti_get(env)->oti_seq_range;
+       struct seq_server_site  *ss = osd_seq_site(osd);
+       int                     rc;
+       ENTRY;
+
+       if (!fid_is_norm(fid) && !fid_is_root(fid))
+               RETURN(0);
+
+       rc = osd_fld_lookup(env, osd, fid, range);
+       if (rc != 0) {
+               CERROR("%s: Can not lookup fld for "DFID"\n",
+                      osd_name(osd), PFID(fid));
+               RETURN(rc);
+       }
+
+       RETURN(ss->ss_node_id != range->lsr_index);
+}
+
 /**
  *      Inserts (key, value) pair in \a directory object.
  *
@@ -499,7 +520,7 @@ static int osd_dir_insert(const struct lu_env *env, struct dt_object *dt,
        struct osd_device   *osd = osd_obj2dev(parent);
        struct lu_fid       *fid = (struct lu_fid *)rec;
        struct osd_thandle  *oh;
-       struct osd_object   *child;
+       struct osd_object   *child = NULL;
        __u32                attr;
        char                *name = (char *)key;
        int                  rc;
@@ -514,46 +535,58 @@ static int osd_dir_insert(const struct lu_env *env, struct dt_object *dt,
        LASSERT(th != NULL);
        oh = container_of0(th, struct osd_thandle, ot_super);
 
-       child = osd_object_find(env, dt, fid);
-       if (IS_ERR(child))
-               RETURN(PTR_ERR(child));
+       rc = osd_remote_fid(env, osd, fid);
+       if (rc < 0) {
+               CERROR("%s: Can not find object "DFID": rc = %d\n",
+                      osd->od_svname, PFID(fid), rc);
+               RETURN(rc);
+       }
+
+       if (unlikely(rc == 1)) {
+               /* Insert remote entry */
+               memset(&oti->oti_zde.lzd_reg, 0, sizeof(oti->oti_zde.lzd_reg));
+               oti->oti_zde.lzd_reg.zde_type = IFTODT(S_IFDIR & S_IFMT);
+       } else {
+               /*
+                * To simulate old Orion setups with ./..  stored in the
+                * directories
+                */
+               /* Insert local entry */
+               child = osd_object_find(env, dt, fid);
+               if (IS_ERR(child))
+                       RETURN(PTR_ERR(child));
 
-/*
- * to simulate old Orion setups with ./..  stored in the directories
- */
 #if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2, 3, 91, 0)
 #define OSD_ZFS_INSERT_DOTS_FOR_TESTING__
 #endif
-
-       LASSERT(child->oo_db);
-       if (name[0] == '.') {
-               if (name[1] == 0) {
-                       /* do not store ".", instead generate it
-                        * during iteration */
+               LASSERT(child->oo_db);
+               if (name[0] == '.') {
+                       if (name[1] == 0) {
+                               /* do not store ".", instead generate it
+                                * during iteration */
 #ifndef OSD_ZFS_INSERT_DOTS_FOR_TESTING
-                       GOTO(out, rc = 0);
+                               GOTO(out, rc = 0);
 #endif
-               } else if (name[1] == '.' && name[2] == 0) {
-                       /* update parent dnode in the child.
-                        * later it will be used to generate ".." */
-                       udmu_objset_t *uos = &osd->od_objset;
-                       rc = osd_object_sa_update(parent, SA_ZPL_PARENT(uos),
-                                                 &child->oo_db->db_object,
-                                                 8, oh);
-
+                       } else if (name[1] == '.' && name[2] == 0) {
+                               /* update parent dnode in the child.
+                                * later it will be used to generate ".." */
+                               udmu_objset_t *uos = &osd->od_objset;
+                               rc = osd_object_sa_update(parent, SA_ZPL_PARENT(uos),
+                                                         &child->oo_db->db_object,
+                                                         8, oh);
 #ifndef OSD_ZFS_INSERT_DOTS_FOR_TESTING
-                       GOTO(out, rc);
+                               GOTO(out, rc);
 #endif
+                       }
                }
+               CLASSERT(sizeof(oti->oti_zde.lzd_reg) == 8);
+               CLASSERT(sizeof(oti->oti_zde) % 8 == 0);
+               attr = child->oo_dt.do_lu.lo_header ->loh_attr;
+               oti->oti_zde.lzd_reg.zde_type = IFTODT(attr & S_IFMT);
+               oti->oti_zde.lzd_reg.zde_dnode = child->oo_db->db_object;
        }
 
-       CLASSERT(sizeof(oti->oti_zde.lzd_reg) == 8);
-       CLASSERT(sizeof(oti->oti_zde) % 8 == 0);
-       attr = child->oo_dt.do_lu.lo_header ->loh_attr;
-       oti->oti_zde.lzd_reg.zde_type = IFTODT(attr & S_IFMT);
-       oti->oti_zde.lzd_reg.zde_dnode = child->oo_db->db_object;
        oti->oti_zde.lzd_fid = *fid;
-
        /* Insert (key,oid) into ZAP */
        rc = -zap_add(osd->od_objset.os, parent->oo_db->db_object,
                      (char *)key, 8, sizeof(oti->oti_zde) / 8,
@@ -562,7 +595,8 @@ static int osd_dir_insert(const struct lu_env *env, struct dt_object *dt,
 #ifndef OSD_ZFS_INSERT_DOTS_FOR_TESTING
 out:
 #endif
-       osd_object_put(env, child);
+       if (child != NULL)
+               osd_object_put(env, child);
 
        RETURN(rc);
 }

diff --git a/lustre/osd-zfs/osd_internal.h b/lustre/osd-zfs/osd_internal.h
index 3801296..9fee6d5 100644 (file)
--- a/lustre/osd-zfs/osd_internal.h
+++ b/lustre/osd-zfs/osd_internal.h
@@ -426,7 +426,8 @@ int osd_convert_root_to_new_seq(const struct lu_env *env,
 /* osd_index.c */
 int osd_index_try(const struct lu_env *env, struct dt_object *dt,
                  const struct dt_index_features *feat);
-
+int osd_fld_lookup(const struct lu_env *env, struct osd_device *osd,
+                  const struct lu_fid *fid, struct lu_seq_range *range);
 
 /* osd_xattr.c */
 int __osd_xattr_load(udmu_objset_t *uos, uint64_t dnode, nvlist_t **sa_xattr);

diff --git a/lustre/osd-zfs/osd_oi.c b/lustre/osd-zfs/osd_oi.c
index 01e6067..9cb4f1f 100644 (file)
--- a/lustre/osd-zfs/osd_oi.c
+++ b/lustre/osd-zfs/osd_oi.c
@@ -219,6 +219,12 @@ int osd_fld_lookup(const struct lu_env *env, struct osd_device *osd,
        struct seq_server_site  *ss = osd_seq_site(osd);
        int                     rc;
 
+       if (fid_is_root(fid)) {
+               range->lsr_flags = LU_SEQ_RANGE_MDT;
+               range->lsr_index = 0;
+               return 0;
+       }
+
        if (fid_is_idif(fid)) {
                range->lsr_flags = LU_SEQ_RANGE_OST;
                range->lsr_index = fid_idif_ost_idx(fid);
@@ -227,6 +233,7 @@ int osd_fld_lookup(const struct lu_env *env, struct osd_device *osd,
 
        if (!fid_is_norm(fid)) {
                range->lsr_flags = LU_SEQ_RANGE_MDT;
+               /* If ss is NULL, it suppose not get lsr_index at all */
                if (ss != NULL)
                        range->lsr_index = ss->ss_node_id;
                return 0;
@@ -235,10 +242,9 @@ int osd_fld_lookup(const struct lu_env *env, struct osd_device *osd,
        LASSERT(ss != NULL);
        range->lsr_flags = -1;
        rc = fld_server_lookup(env, ss->ss_server_fld, fid_seq(fid), range);
-       if (rc != 0) {
+       if (rc != 0)
                CERROR("%s can not find "DFID": rc = %d\n",
-                      osd2lu_dev(osd)->ld_obd->obd_name, PFID(fid), rc);
-       }
+                      osd_name(osd), PFID(fid), rc);
        return rc;
 }