LU-5099 api: transfer object type via dt_insert API

[fs/lustre-release.git] / lustre / osd-zfs / osd_index.c

diff --git a/lustre/osd-zfs/osd_index.c b/lustre/osd-zfs/osd_index.c
index b96e642..43779a9 100644 (file)
--- a/lustre/osd-zfs/osd_index.c
+++ b/lustre/osd-zfs/osd_index.c
@@ -480,8 +480,8 @@ static int osd_seq_exists(const struct lu_env *env, struct osd_device *osd,
        int                     rc;
        ENTRY;
 
-       if (ss == NULL)
-               RETURN(1);
+       LASSERT(ss != NULL);
+       LASSERT(ss->ss_server_fld != NULL);
 
        rc = osd_fld_lookup(env, osd, seq, range);
        if (rc != 0) {
@@ -494,14 +494,22 @@ static int osd_seq_exists(const struct lu_env *env, struct osd_device *osd,
 }
 
 static int osd_remote_fid(const struct lu_env *env, struct osd_device *osd,
-                         struct lu_fid *fid)
+                         const struct lu_fid *fid)
 {
+       struct seq_server_site  *ss = osd_seq_site(osd);
        ENTRY;
 
        /* FID seqs not in FLDB, must be local seq */
        if (unlikely(!fid_seq_in_fldb(fid_seq(fid))))
                RETURN(0);
 
+       /* If FLD is not being initialized yet, it only happens during the
+        * initialization, likely during mgs initialization, and we assume
+        * this is local FID. */
+       if (ss == NULL || ss->ss_server_fld == NULL)
+               RETURN(0);
+
+       /* Only check the local FLDB here */
        if (osd_seq_exists(env, osd, fid_seq(fid)))
                RETURN(0);
 
@@ -529,7 +537,8 @@ static int osd_dir_insert(const struct lu_env *env, struct dt_object *dt,
        struct osd_thread_info *oti = osd_oti_get(env);
        struct osd_object   *parent = osd_dt_obj(dt);
        struct osd_device   *osd = osd_obj2dev(parent);
-       struct lu_fid       *fid = (struct lu_fid *)rec;
+       struct dt_insert_rec *rec1 = (struct dt_insert_rec *)rec;
+       const struct lu_fid *fid = rec1->rec_fid;
        struct osd_thandle  *oh;
        struct osd_object   *child = NULL;
        __u32                attr;
@@ -556,7 +565,7 @@ static int osd_dir_insert(const struct lu_env *env, struct dt_object *dt,
        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);
+               oti->oti_zde.lzd_reg.zde_type = IFTODT(rec1->rec_type & S_IFMT);
        } else {
                /*
                 * To simulate old Orion setups with ./..  stored in the
@@ -969,6 +978,44 @@ out:
        RETURN(rc);
 }
 
+static int osd_dir_it_rec_size(const struct lu_env *env, const struct dt_it *di,
+                              __u32 attr)
+{
+       struct osd_zap_it   *it = (struct osd_zap_it *)di;
+       zap_attribute_t     *za = &osd_oti_get(env)->oti_za;
+       int                  rc, namelen = 0;
+       ENTRY;
+
+       if (it->ozi_pos <= 1)
+               namelen = 1;
+       else if (it->ozi_pos == 2)
+               namelen = 2;
+
+       if (namelen > 0) {
+               rc = lu_dirent_calc_size(namelen, attr);
+               RETURN(rc);
+       }
+
+       rc = -zap_cursor_retrieve(it->ozi_zc, za);
+       if (unlikely(rc != 0))
+               RETURN(rc);
+
+       if (za->za_integer_length != 8 || za->za_num_integers < 3) {
+               CERROR("%s: unsupported direntry format: %d %d\n",
+                      osd_obj2dev(it->ozi_obj)->od_svname,
+                      za->za_integer_length, (int)za->za_num_integers);
+               RETURN(-EIO);
+       }
+
+       namelen = strlen(za->za_name);
+       if (namelen > NAME_MAX)
+               RETURN(-EOVERFLOW);
+
+       rc = lu_dirent_calc_size(namelen, attr);
+
+       RETURN(rc);
+}
+
 static __u64 osd_dir_it_store(const struct lu_env *env, const struct dt_it *di)
 {
        struct osd_zap_it *it = (struct osd_zap_it *)di;
@@ -1035,6 +1082,7 @@ static struct dt_index_operations osd_dir_ops = {
                .key      = osd_dir_it_key,
                .key_size = osd_dir_it_key_size,
                .rec      = osd_dir_it_rec,
+               .rec_size = osd_dir_it_rec_size,
                .store    = osd_dir_it_store,
                .load     = osd_dir_it_load
        }
@@ -1195,7 +1243,8 @@ static int osd_index_it_get(const struct lu_env *env, struct dt_it *di,
         * XXX: we need a binary version of zap_cursor_move_to_key()
         *      to implement this API */
        if (*((const __u64 *)key) != 0)
-               CERROR("NOT IMPLEMETED YET (move to %Lx)\n", *((__u64 *)key));
+               CERROR("NOT IMPLEMETED YET (move to "LPX64")\n",
+                      *((__u64 *)key));
 
        zap_cursor_fini(it->ozi_zc);
        memset(it->ozi_zc, 0, sizeof(*it->ozi_zc));
@@ -1360,6 +1409,11 @@ static struct dt_it *osd_zfs_otable_it_init(const struct lu_env *env,
 
        it->mit_dev = dev;
 
+       /* XXX: dmu_object_next() does NOT find dnodes allocated
+        *      in the current non-committed txg, so we force txg
+        *      commit to find all existing dnodes ... */
+       txg_wait_synced(dmu_objset_pool(dev->od_objset.os), 0ULL);
+
        RETURN((struct dt_it *)it);
 }
 
@@ -1425,7 +1479,6 @@ static int osd_zfs_otable_it_next(const struct lu_env *env, struct dt_it *di)
        uchar_t                 *v;
        __u64                    dnode;
        int                      rc, s;
-       ENTRY;
 
        memset(&it->mit_fid, 0, sizeof(it->mit_fid));
 
@@ -1473,7 +1526,7 @@ static int osd_zfs_otable_it_next(const struct lu_env *env, struct dt_it *di)
               it->mit_pos, PFID(&it->mit_fid), rc);
 
 out:
-       RETURN(rc);
+       return rc;
 }
 
 static struct dt_key *osd_zfs_otable_it_key(const struct lu_env *env,