- on-disk compatibility in ldiskfs osd:

  - hard-coded fids are used to access files like last_rcvd, LAST_ID, LAST_GROUP
  - osd recognizes idif and uses legacy OST hierarchy instead of OI
- ofd uses hard-coded fids to access files like last_rcvd, LAST_ID, LAST_GROUP
- minor cleanups in ofd

diff --git a/lustre/include/lustre_fid.h b/lustre/include/lustre_fid.h
index 921b423..f28ff48 100644 (file)
--- a/lustre/include/lustre_fid.h
+++ b/lustre/include/lustre_fid.h
@@ -107,6 +107,10 @@ enum {
         MDT_LAST_RECV_OID       = 11UL,
         /** \see osd_mod_init */
         OSD_REM_OBJ_DIR_OID     = 12UL,
+        OFD_LAST_RECV_OID       = 19UL,
+        OFD_GROUP0_LAST_ID      = 20UL,
+        OFD_GROUP4K_LAST_ID     = 20UL+4096,
+        OFD_LAST_GROUP          = 4117UL
 };
 
 static inline void lu_local_obj_fid(struct lu_fid *fid, __u32 oid)

diff --git a/lustre/ofd/ofd_fs.c b/lustre/ofd/ofd_fs.c
index 1c99273..49d667e 100644 (file)
--- a/lustre/ofd/ofd_fs.c
+++ b/lustre/ofd/ofd_fs.c
@@ -45,6 +45,47 @@
 
 #include "ofd_internal.h"
 
+int filter_record_write(const struct lu_env *env, struct filter_device *ofd,
+                        struct dt_object *dt, void *data, loff_t _off, int len,
+                        struct thandle *_th)
+{
+        struct thandle *th;
+        struct lu_buf   buf;
+        loff_t          off;
+        int             rc;
+        ENTRY;
+
+        LASSERT(ofd);
+        LASSERT(dt);
+        LASSERT(len);
+        LASSERT(data);
+
+        buf.lb_buf = data;
+        buf.lb_len = len;
+        off = _off;
+
+        th = _th;
+        if (th == NULL) {
+                th = filter_trans_create(env, ofd);
+                if (IS_ERR(th))
+                        RETURN(PTR_ERR(th));
+                rc = dt_declare_record_write(env, dt, buf.lb_len, off, th);
+                LASSERT(rc == 0);
+                rc = filter_trans_start(env, ofd, th);
+                if (rc)
+                        RETURN(rc);
+        }
+
+        rc = dt_record_write(env, dt, &buf, &off, th);
+
+        if (_th == NULL) {
+                LASSERT(th);
+                filter_trans_stop(env, ofd, th);
+        }
+
+        RETURN(rc);
+}
+
 obd_id filter_last_id(struct filter_device *ofd, obd_gr group)
 {
         obd_id id;
@@ -67,123 +108,154 @@ void filter_last_id_set(struct filter_device *ofd, obd_id id, obd_gr group)
         spin_unlock(&ofd->ofd_objid_lock);
 }
 
-int filter_last_id_read(const struct lu_env *env, struct filter_device *ofd,
-                        obd_gr group)
+int filter_last_id_write(const struct lu_env *env, struct filter_device *ofd,
+                         obd_gr group, struct thandle *th)
 {
-        struct filter_thread_info *info = filter_info(env);
-        struct lu_buf              buf;
-        loff_t                     off;
-        obd_id tmp;
-        int rc;
+        obd_id          tmp;
+        int             rc;
         ENTRY;
 
-        LASSERT(ofd->ofd_groups_file != NULL);
-        LASSERT(info);
+        CDEBUG(D_INODE, "%s: write last_objid for group "LPU64": "LPU64"\n",
+               filter_obd(ofd)->obd_name, group, filter_last_id(ofd, group));
 
-        buf.lb_buf = &tmp;
-        buf.lb_len = sizeof(tmp);
-        off = group * sizeof(tmp);
+        tmp = cpu_to_le64(filter_last_id(ofd, group));
 
-        rc = dt_record_read(env, ofd->ofd_groups_file, &buf, &off);
-        if (rc >= 0) {
-                filter_last_id_set(ofd, le64_to_cpu(tmp), group);
-                CDEBUG(D_INODE, "%s: read last_objid for group "LPU64": "
-                       LPU64"\n", filter_obd(ofd)->obd_name, group,
-                       filter_last_id(ofd, group));
-        } else
-                CERROR("read group "LPU64" last objid: rc = %d\n", group, rc);
+        rc = filter_record_write(env, ofd, ofd->ofd_lastid_obj[group],
+                                 &tmp, 0, sizeof(tmp), th);
 
         RETURN(rc);
 }
 
-int filter_last_id_write(const struct lu_env *env, struct filter_device *ofd,
-                         obd_gr group, int force_sync)
+int filter_last_group_write(const struct lu_env *env, struct filter_device *ofd)
 {
-        struct thandle *th;
-        struct lu_buf   buf;
-        obd_id          tmp;
-        loff_t          off;
-        int             rc;
+        __u32   tmp;
+        int     rc;
         ENTRY;
 
-        CDEBUG(D_INODE, "%s: write last_objid for group "LPU64": "LPU64"\n",
-               filter_obd(ofd)->obd_name, group, filter_last_id(ofd, group));
+        tmp = cpu_to_le32(ofd->ofd_max_group);
 
-        LASSERT(ofd->ofd_groups_file != NULL);
+        rc = filter_record_write(env, ofd, ofd->ofd_last_group_file,
+                                 &tmp, 0, sizeof(tmp), NULL);
 
-        tmp = cpu_to_le64(filter_last_id(ofd, group));
-        buf.lb_buf = &tmp;
-        buf.lb_len = sizeof(tmp);
-        off = group * sizeof(tmp);
-
-        th = filter_trans_create(env, ofd);
-        if (IS_ERR(th))
-                RETURN(PTR_ERR(th));
-        rc = dt_declare_record_write(env, ofd->ofd_groups_file,
-                                     buf.lb_len, off, th);
-        LASSERT(rc == 0);
-        rc = filter_trans_start(env, ofd, th);
-        if (rc)
-                RETURN(rc);
+        RETURN(rc);
+}
+
+int filter_group_load(const struct lu_env *env,
+                      struct filter_device *ofd, int group)
+{
+        struct filter_object      *fo;
+        struct dt_object          *dob;
+        struct lu_fid              fid;
+        struct lu_attr             attr;
+        struct lu_buf              buf;
+        loff_t                     off;
+        __u64                      lastid;
+        int                        rc;
+        ENTRY;
+
+        /* if group is already initialized */
+        if (ofd->ofd_lastid_obj[group])
+                GOTO(cleanup, rc = 0);
+
+        lu_local_obj_fid(&fid, OFD_GROUP0_LAST_ID + group);
+        memset(&attr, 0, sizeof(attr));
+        attr.la_valid = LA_MODE;
+        attr.la_mode = S_IFREG | 0666;
 
-        rc = dt_record_write(env, ofd->ofd_groups_file, &buf, &off, th);
+        /* create object tracking per-group last created
+         * id to be used by orphan recovery mechanism */
+        fo = filter_object_find_or_create(env, ofd, &fid, &attr);
+        LASSERT(!IS_ERR(fo));
+        dob = filter_object_child(fo);
+        ofd->ofd_lastid_obj[group] = dob;
+
+        rc = dt_attr_get(env, dob, &attr, BYPASS_CAPA);
         if (rc)
-                CERROR("write group "LPU64" last objid: rc = %d\n", group, rc);
+                GOTO(cleanup, rc);
+
+        if (attr.la_size == 0) {
+                /* object is just created, initialize last id */
+                ofd->ofd_last_objids[group] = FILTER_INIT_OBJID;
+                filter_last_id_write(env, ofd, group, 0);
+
+        } else if (attr.la_size == sizeof(lastid)) {
+                off = 0;
+                buf.lb_buf = &lastid;
+                buf.lb_len = sizeof(lastid);
+                rc = dt_record_read(env, dob, &buf, &off);
+                if (rc) {
+                        CERROR("can't read last_id: %d\n", rc);
+                        GOTO(cleanup, rc);
+                }
+                ofd->ofd_last_objids[group] = le64_to_cpu(lastid);
 
-        filter_trans_stop(env, ofd, th);
+        } else {
+                CERROR("corrupted size %Lu LAST_ID of group %u\n",
+                       (unsigned long long) attr.la_size, group);
+                rc = -EINVAL;
+        }
 
+cleanup:
         RETURN(rc);
 }
-
+        
 /* filter groups managements */
 int filter_groups_init(const struct lu_env *env, struct filter_device *ofd)
 {
         struct filter_thread_info *info = filter_info(env);
-        unsigned long groups_size;
-        struct lu_buf buf;
-        loff_t        off;
-        obd_id lastid;
-        int rc, i;
+        unsigned long              groups_size;
+        __u32                      last_group;
+        struct lu_buf              buf;
+        loff_t                     off;
+        int rc = 0, i;
+        ENTRY;
 
         spin_lock_init(&ofd->ofd_objid_lock);
 
-        rc = dt_attr_get(env, ofd->ofd_groups_file, &info->fti_attr,
-                         BYPASS_CAPA);
+        rc = dt_attr_get(env, ofd->ofd_last_group_file,
+                         &info->fti_attr, BYPASS_CAPA);
         if (rc)
-                RETURN(rc);
+                GOTO(cleanup, rc);
 
         groups_size = (unsigned long)info->fti_attr.la_size;
 
         if (groups_size == 0) {
                 CWARN("%s: no groups yet\n", filter_obd(ofd)->obd_name);
-                RETURN(0);
+                ofd->ofd_max_group = 0;
+                goto skip_read;
         }
 
-        if (groups_size % sizeof(lastid) != 0) {
+        if (groups_size != sizeof(last_group)) {
                 CERROR("groups file is corrupted? size = %lu\n", groups_size);
-                RETURN(-EIO);
+                GOTO(cleanup, rc = -EIO);
         }
 
-        ofd->ofd_max_group = groups_size / sizeof(lastid);
-        LASSERT(ofd->ofd_max_group <= FILTER_MAX_GROUPS); /* XXX: dynamic? */
-
         off = 0;
-        buf.lb_buf = &ofd->ofd_last_objids;
-        buf.lb_len = sizeof(lastid) * ofd->ofd_max_group;
-        rc = dt_record_read(env, ofd->ofd_groups_file, &buf, &off);
+        buf.lb_buf = &last_group;
+        buf.lb_len = sizeof(last_group);
+        rc = dt_record_read(env, ofd->ofd_last_group_file, &buf, &off);
         if (rc) {
-                CERROR("can't initialize last_ids: %d\n", rc);
-                RETURN(rc);
+                CERROR("can't read LAST_GROUP: %d\n", rc);
+                GOTO(cleanup, rc);
         }
-        /* last objids are stored in le format, convert them to cpu */
-        spin_lock(&ofd->ofd_objid_lock);
-        for (i = 0; i < groups_size; i++)
-                ofd->ofd_last_objids[i] = le64_to_cpu(ofd->ofd_last_objids[i]);
-        spin_unlock(&ofd->ofd_objid_lock);
 
-        CWARN("%s: %u groups initialized\n", filter_obd(ofd)->obd_name,
-              ofd->ofd_max_group);
-        RETURN(0);
+        ofd->ofd_max_group = le32_to_cpu(last_group);
+        LASSERT(ofd->ofd_max_group <= FILTER_MAX_GROUPS); /* XXX: dynamic? */
+
+skip_read:
+        for (i = 0; i <= ofd->ofd_max_group; i++) {
+               rc = filter_group_load(env, ofd, i);
+               if (rc) {
+                        CERROR("can't load group %d: %d\n", i, rc);
+                        GOTO(cleanup, rc);
+               }
+        }
+
+        CWARN("%s: %u groups initialized\n",
+              filter_obd(ofd)->obd_name, ofd->ofd_max_group);
+
+cleanup:
+        RETURN(rc);
 }
 
 static inline void fsd_le_to_cpu(struct lr_server_data *buf,
@@ -253,21 +325,17 @@ int filter_last_rcvd_header_write(const struct lu_env *env,
                                   struct thandle *th)
 {
         struct filter_thread_info *info;
-        struct lu_buf              buf;
-        loff_t                     off;
         int                        rc;
         ENTRY;
 
         info = lu_context_key_get(&env->le_ctx, &filter_thread_key);
         LASSERT(info);
 
-        buf.lb_buf = &info->fti_fsd;
-        buf.lb_len = sizeof(info->fti_fsd);
-        off = 0;
-
         fsd_cpu_to_le(&ofd->ofd_fsd, &info->fti_fsd);
 
-        rc = dt_record_write(env, ofd->ofd_last_rcvd, &buf, &off, th);
+        rc = filter_record_write(env, ofd, ofd->ofd_last_rcvd,
+                                 &info->fti_fsd, 0, sizeof(info->fti_fsd), th);
+
         CDEBUG(D_INFO, "write last_rcvd header rc = %d:\n"
                "uuid = %s\nlast_transno = "LPU64"\n",
                rc, ofd->ofd_fsd.lsd_uuid, ofd->ofd_fsd.lsd_last_transno);
@@ -305,8 +373,10 @@ int filter_last_rcvd_write(const struct lu_env *env,
 
         buf.lb_buf = &info->fti_fsd;
         buf.lb_len = sizeof(info->fti_fsd);
+        
+        rc = filter_record_write(env, ofd, ofd->ofd_last_rcvd,
+                                 &info->fti_fsd, 0, sizeof(info->fti_fsd), th);
 
-        rc = dt_record_write(env, ofd->ofd_last_rcvd, &buf, off, th);
         return rc;
 }
 
@@ -429,20 +499,8 @@ int filter_server_data_update(const struct lu_env *env,
          * This may be called from difficult reply handler and
          * mdt->mdt_last_rcvd may be NULL that time.
          */
-        if (ofd->ofd_last_rcvd != NULL) {
-                struct thandle *th;
-
-                th = filter_trans_create0(env, ofd);
-                if (IS_ERR(th))
-                        RETURN(PTR_ERR(th));
-                dt_declare_record_write(env, ofd->ofd_last_rcvd,
-                                        sizeof(ofd->ofd_fsd), 0, th);
-                rc = filter_trans_start(env, ofd, th);
-                if (rc)
-                        RETURN(rc);
-                rc = filter_last_rcvd_header_write(env, ofd, th);
-                filter_trans_stop(env, ofd, th);
-        }
+        if (ofd->ofd_last_rcvd != NULL)
+                rc = filter_last_rcvd_header_write(env, ofd, NULL);
 
         RETURN(rc);
 }

diff --git a/lustre/ofd/ofd_internal.h b/lustre/ofd/ofd_internal.h
index 3c29493..6b093cb 100644 (file)
--- a/lustre/ofd/ofd_internal.h
+++ b/lustre/ofd/ofd_internal.h
@@ -138,7 +138,7 @@ struct filter_device {
 
         /* last_rcvd file */
         struct lu_target         ofd_lut;
-        struct dt_object        *ofd_groups_file;
+        struct dt_object        *ofd_last_group_file;
         unsigned long           *ofd_last_rcvd_slots;
 
         int                      ofd_subdir_count;
@@ -151,6 +151,7 @@ struct filter_device {
         int                      ofd_max_group;
         obd_id                   ofd_last_objids[FILTER_MAX_GROUPS];
         struct semaphore         ofd_create_locks[FILTER_MAX_GROUPS];
+        struct dt_object        *ofd_lastid_obj[FILTER_MAX_GROUPS];
         spinlock_t               ofd_objid_lock;
         unsigned long            ofd_destroys_in_progress;
 
@@ -447,7 +448,7 @@ void filter_fs_cleanup(const struct lu_env *env, struct filter_device *ofd);
 obd_id filter_last_id(struct filter_device *ofd, obd_gr group);
 void filter_last_id_set(struct filter_device *ofd, obd_id id, obd_gr group);
 int filter_last_id_write(const struct lu_env *env, struct filter_device *ofd,
-                         obd_gr group, int force_sync);
+                         obd_gr group, struct thandle *th);
 int filter_last_id_read(const struct lu_env *env, struct filter_device *ofd,
                         obd_gr group);
 int filter_groups_init(const struct lu_env *env, struct filter_device *ofd);
@@ -462,6 +463,9 @@ int filter_server_data_init(const struct lu_env *env,
                             struct filter_device *ofd);
 int filter_server_data_update(const struct lu_env *env,
                               struct filter_device *ofd);
+int filter_group_load(const struct lu_env *env,
+                      struct filter_device *ofd, int group);
+int filter_last_group_write(const struct lu_env *env, struct filter_device *ofd);
 
 /* filter_objects.c */
 struct filter_object *filter_object_find(const struct lu_env *env,
@@ -508,7 +512,7 @@ void filter_grant_commit(struct obd_export *exp, int niocount,
 static inline void lu_idif_build(struct lu_fid *fid, obd_id id, obd_gr gr)
 {
         LASSERT((id >> 48) == 0);
-        fid->f_seq = (0x200000000ULL | id >> 32);
+        fid->f_seq = (IDIF_SEQ_START| id >> 32);
         fid->f_oid = (__u32)(id & 0xffffffff);
         fid->f_ver = gr;
 }

diff --git a/lustre/ofd/ofd_obd.c b/lustre/ofd/ofd_obd.c
index d635961..e64a739 100644 (file)
--- a/lustre/ofd/ofd_obd.c
+++ b/lustre/ofd/ofd_obd.c
@@ -263,7 +263,9 @@ static int filter_obd_connect(const struct lu_env *env, struct obd_export **_exp
         if (group > ofd->ofd_max_group) {
                 ofd->ofd_max_group = group;
                 filter_last_id_set(ofd, FILTER_INIT_OBJID, group);
-                filter_last_id_write(env, ofd, group, 1);
+                filter_last_id_write(env, ofd, group, NULL);
+                filter_last_group_write(env, ofd);
+                rc = filter_group_load(env, ofd, group);
         }
 
 out:
@@ -921,7 +923,7 @@ static int filter_orphans_destroy(const struct lu_env *env,
         CDEBUG(D_HA, "%s: after destroy: set last_objids["LPU64"] = "LPU64"\n",
                filter_obd(ofd)->obd_name, gr, mds_id);
         if (!skip_orphan) {
-                rc = filter_last_id_write(env, ofd, gr, 1);
+                rc = filter_last_id_write(env, ofd, gr, NULL);
         } else {
                 /* don't reuse orphan object, return last used objid */
                 oa->o_id = last;

diff --git a/lustre/ofd/ofd_objects.c b/lustre/ofd/ofd_objects.c
index 4cf187a..58d96a6 100644 (file)
--- a/lustre/ofd/ofd_objects.c
+++ b/lustre/ofd/ofd_objects.c
@@ -164,7 +164,7 @@ int filter_precreate_object(const struct lu_env *env, struct filter_device *ofd,
         if (rc)
                 GOTO(trans_stop, rc);
 
-        rc = dt_declare_record_write(env, ofd->ofd_groups_file,
+        rc = dt_declare_record_write(env, ofd->ofd_lastid_obj[group],
                                      sizeof(tmp), off, th);
         if (rc)
                 GOTO(trans_stop, rc);
@@ -190,9 +190,8 @@ int filter_precreate_object(const struct lu_env *env, struct filter_device *ofd,
         LASSERT(filter_object_exists(fo));
 
         filter_last_id_set(ofd, id, group);
-        tmp = cpu_to_le64(filter_last_id(ofd, group));
 
-        rc = dt_record_write(env, ofd->ofd_groups_file, &buf, &off, th);
+        rc = filter_last_id_write(env, ofd, group, th);
 
 out_unlock:
         filter_write_unlock(env, fo);

diff --git a/lustre/ofd/ofd_recovery.c b/lustre/ofd/ofd_recovery.c
index 6f9e45c..0826549 100644 (file)
--- a/lustre/ofd/ofd_recovery.c
+++ b/lustre/ofd/ofd_recovery.c
@@ -273,7 +273,7 @@ int filter_fs_setup(const struct lu_env *env, struct filter_device *ofd,
 
         dt_txn_callback_add(ofd->ofd_osd, &ofd->ofd_txn_cb);
 
-        lu_local_obj_fid(&fid, MDT_LAST_RECV_OID);
+        lu_local_obj_fid(&fid, OFD_LAST_RECV_OID);
         memset(&attr, 0, sizeof(attr));
         attr.la_valid = LA_MODE;
         attr.la_mode = S_IFREG | 0666;
@@ -284,14 +284,14 @@ int filter_fs_setup(const struct lu_env *env, struct filter_device *ofd,
         rc = filter_server_data_init(env, ofd);
         LASSERT(rc == 0);
 
-        lu_local_obj_fid(&fid, MDD_OBJECTS_OID);
+        lu_local_obj_fid(&fid, OFD_LAST_GROUP);
         memset(&attr, 0, sizeof(attr));
         attr.la_valid = LA_MODE;
         attr.la_mode = S_IFREG | 0666;
 
         fo = filter_object_find_or_create(env, ofd, &fid, &attr);
         LASSERT(!IS_ERR(fo));
-        ofd->ofd_groups_file = filter_object_child(fo);
+        ofd->ofd_last_group_file = filter_object_child(fo);
         rc = filter_groups_init(env, ofd);
         LASSERT(rc == 0);
 
@@ -316,15 +316,18 @@ void filter_fs_cleanup(const struct lu_env *env, struct filter_device *ofd)
 
         filter_server_data_update(env, ofd);
 
-        for (i = 0; i < ofd->ofd_max_group; i++)
-                filter_last_id_write(env, ofd, i, (i == ofd->ofd_max_group-1));
+        for (i = 0; i <= ofd->ofd_max_group; i++) {
+                filter_last_id_write(env, ofd, i, NULL);
+                if (ofd->ofd_lastid_obj[i])
+                        lu_object_put(env, &ofd->ofd_lastid_obj[i]->do_lu);
+        }
 
         /* Remove transaction callback */
         dt_txn_callback_del(ofd->ofd_osd, &ofd->ofd_txn_cb);
 
-        if (ofd->ofd_groups_file)
-                lu_object_put(env, &ofd->ofd_groups_file->do_lu);
-        ofd->ofd_groups_file = NULL;
+        if (ofd->ofd_last_group_file)
+                lu_object_put(env, &ofd->ofd_last_group_file->do_lu);
+        ofd->ofd_last_group_file = NULL;
 
         OBD_FREE(ofd->ofd_last_rcvd_slots, LR_MAX_CLIENTS / 8);
 

diff --git a/lustre/osd/Makefile.in b/lustre/osd/Makefile.in
index 0f6dc11..6769f8a 100644 (file)
--- a/lustre/osd/Makefile.in
+++ b/lustre/osd/Makefile.in
@@ -1,5 +1,5 @@
 MODULES := osd
-osd-objs := osd_handler.o osd_oi.o osd_igif.o osd_lproc.o osd_io.o
+osd-objs := osd_handler.o osd_oi.o osd_igif.o osd_lproc.o osd_io.o osd_compat.o
 
 EXTRA_PRE_CFLAGS := -I@LINUX@/fs -I@LDISKFS_DIR@ -I@LDISKFS_DIR@/ldiskfs
 

diff --git a/lustre/osd/osd_handler.c b/lustre/osd/osd_handler.c
index 4677930..38ed948 100644 (file)
--- a/lustre/osd/osd_handler.c
+++ b/lustre/osd/osd_handler.c
@@ -504,7 +504,7 @@ static int osd_inode_remove(const struct lu_env *env, struct osd_object *obj)
 
         result = osd_trans_start(env, &osd->od_dt_dev, th);
         if (result == 0)
-                result = osd_oi_delete(oti, &osd->od_oi, fid, th);
+                result = osd_oi_delete(oti, osd, fid, th);
         osd_trans_stop(env, th);
         return result;
 }
@@ -1813,7 +1813,7 @@ static int __osd_oi_insert(const struct lu_env *env, struct osd_object *obj,
         id->oii_ino = obj->oo_inode->i_ino;
         id->oii_gen = obj->oo_inode->i_generation;
 
-        return osd_oi_insert(info, &osd->od_oi, fid, id, th,
+        return osd_oi_insert(info, osd, fid, id, th,
                              uc ? uc->mu_cap & CFS_CAP_SYS_RESOURCE_MASK : 1);
 }
 
@@ -3885,6 +3885,8 @@ static struct lu_device *osd_device_fini(const struct lu_env *env,
         int rc;
         ENTRY;
 
+        osd_compat_fini(osd_dev(d));
+
         if (osd_dev(d)->od_mnt) {
                 shrink_dcache_sb(osd_sb(osd_dev(d)));
                 osd_sync(env, lu2dt_dev(d));
@@ -4007,6 +4009,10 @@ static int osd_prepare(const struct lu_env *env,
         if (result != 0)
                 RETURN(result);
 
+        result = osd_compat_init(osd);
+        if (result != 0)
+                RETURN(result);
+
         if (lu_device_is_md(pdev)) {
                 /* 2. setup local objects */
                 result = llo_local_objects_setup(env, lu2md_dev(pdev),
@@ -4068,13 +4074,12 @@ static int osd_fid_lookup(const struct lu_env *env,
         struct lu_device       *ldev = obj->oo_dt.do_lu.lo_dev;
         struct osd_device      *dev;
         struct osd_inode_id    *id;
-        struct osd_oi          *oi;
         struct inode           *inode;
         int                     result;
 
         LINVRNT(osd_invariant(obj));
         LASSERT(obj->oo_inode == NULL);
-        LASSERT(fid_is_sane(fid));
+        LASSERT(fid_is_sane(fid) || fid_is_idif(fid));
         /*
          * This assertion checks that osd layer sees only local
          * fids. Unfortunately it is somewhat expensive (does a
@@ -4087,12 +4092,11 @@ static int osd_fid_lookup(const struct lu_env *env,
         info = osd_oti_get(env);
         dev  = osd_dev(ldev);
         id   = &info->oti_id;
-        oi   = &dev->od_oi;
 
         if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOENT))
                 RETURN(-ENOENT);
 
-        result = osd_oi_lookup(info, oi, fid, id);
+        result = osd_oi_lookup(info, dev, fid, id);
         if (result == 0) {
                 inode = osd_iget(info, dev, id);
                 if (!IS_ERR(inode)) {

diff --git a/lustre/osd/osd_internal.h b/lustre/osd/osd_internal.h
index 11276dc..e16ac0c 100644 (file)
--- a/lustre/osd/osd_internal.h
+++ b/lustre/osd/osd_internal.h
@@ -176,6 +176,8 @@ struct osd_object {
 #endif
 };
 
+struct osd_compat_objid;
+
 /*
  * osd device.
  */
@@ -226,6 +228,8 @@ struct osd_device {
         __u32                     od_iop_mode;
 
         struct fsfilt_operations *od_fsops;
+
+        struct osd_compat_objid  *od_ost_map;
 };
 
 struct osd_it_ea_dirent {
@@ -356,6 +360,9 @@ struct osd_thread_info {
 
         /** 0-copy IO */
         struct filter_iobuf    oti_iobuf;
+
+        /** used by compat stuff */
+        struct inode           oti_inode;
 };
 
 #ifdef LPROCFS
@@ -389,5 +396,20 @@ extern struct buffer_head * ldiskfs_find_entry(struct dentry *dentry,
                                                struct ldiskfs_dir_entry_2
                                                ** res_dir);
 
+int osd_compat_init(struct osd_device *osd);
+void osd_compat_fini(const struct osd_device *dev);
+int osd_compat_objid_lookup(struct osd_thread_info *info, struct osd_device *osd,
+                            const struct lu_fid *fid, struct osd_inode_id *id);
+int osd_compat_objid_insert(struct osd_thread_info *info, struct osd_device *osd,
+                            const struct lu_fid *fid, const struct osd_inode_id *id,
+                            struct thandle *th);
+int osd_compat_objid_delete(struct osd_thread_info *info, struct osd_device *osd,
+                            const struct lu_fid *fid, struct thandle *th);
+int osd_compat_spec_lookup(struct osd_thread_info *info, struct osd_device *osd,
+                           const struct lu_fid *fid, struct osd_inode_id *id);
+int osd_compat_spec_insert(struct osd_thread_info *info, struct osd_device *osd,
+                           const struct lu_fid *fid, const struct osd_inode_id *id,
+                           struct thandle *th);
+
 #endif /* __KERNEL__ */
 #endif /* _OSD_INTERNAL_H */

diff --git a/lustre/osd/osd_oi.c b/lustre/osd/osd_oi.c
index 22a1348..0ccb058 100644 (file)
--- a/lustre/osd/osd_oi.c
+++ b/lustre/osd/osd_oi.c
@@ -297,12 +297,16 @@ static inline int fid_is_oi_fid(const struct lu_fid *fid)
                 fid_oid(fid) == OSD_OI_FID_OTHER_OID)));
 }
 
-int osd_oi_lookup(struct osd_thread_info *info, struct osd_oi *oi,
+int osd_oi_lookup(struct osd_thread_info *info, struct osd_device *osd,
                   const struct lu_fid *fid, struct osd_inode_id *id)
 {
+        struct osd_oi *oi = &osd->od_oi;
         int rc;
 
-        if (fid_is_igif(fid)) {
+        if (fid_is_idif(fid)) {
+                /* old OSD obj id */
+                rc = osd_compat_objid_lookup(info, osd, fid, id);
+        } else if (fid_is_igif(fid)) {
                 lu_igif_to_id(fid, id);
                 rc = 0;
         } else {
@@ -329,7 +333,7 @@ int osd_oi_lookup(struct osd_thread_info *info, struct osd_oi *oi,
         return rc;
 }
 
-int osd_oi_insert(struct osd_thread_info *info, struct osd_oi *oi,
+int osd_oi_insert(struct osd_thread_info *info, struct osd_device *osd,
                   const struct lu_fid *fid, const struct osd_inode_id *id0,
                   struct thandle *th, int ignore_quota)
 {
@@ -343,10 +347,18 @@ int osd_oi_insert(struct osd_thread_info *info, struct osd_oi *oi,
         if (fid_is_oi_fid(fid))
                 return 0;
 
-        key = oi_fid_key(info, oi, fid, &idx);
+        if (fid_is_idif(fid))
+                return osd_compat_objid_insert(info, osd, fid, id0, th);
+
+        /* notice we don't return immediately, but continue to get into OI */
+        if (unlikely(fid_seq(fid) == FID_SEQ_LOCAL_FILE))
+                osd_compat_spec_insert(info, osd, fid, id0, th);
+
+        key = oi_fid_key(info, &osd->od_oi, fid, &idx);
         id  = &info->oti_id;
         id->oii_ino = cpu_to_be32(id0->oii_ino);
         id->oii_gen = cpu_to_be32(id0->oii_gen);
+
         return idx->do_index_ops->dio_insert(info->oti_env, idx,
                                              (const struct dt_rec *)id,
                                              key, th, BYPASS_CAPA,
@@ -354,7 +366,7 @@ int osd_oi_insert(struct osd_thread_info *info, struct osd_oi *oi,
 }
 
 int osd_oi_delete(struct osd_thread_info *info,
-                  struct osd_oi *oi, const struct lu_fid *fid,
+                  struct osd_device *osd, const struct lu_fid *fid,
                   struct thandle *th)
 {
         struct dt_object    *idx;
@@ -363,7 +375,12 @@ int osd_oi_delete(struct osd_thread_info *info,
         if (fid_is_igif(fid))
                 return 0;
 
-        key = oi_fid_key(info, oi, fid, &idx);
+        LASSERT(fid_seq(fid) != FID_SEQ_LOCAL_FILE);
+
+        if (fid_is_idif(fid))
+                return osd_compat_objid_delete(info, osd, fid, th);
+
+        key = oi_fid_key(info, &osd->od_oi, fid, &idx);
         return idx->do_index_ops->dio_delete(info->oti_env, idx,
                                              key, th, BYPASS_CAPA);
 }

diff --git a/lustre/osd/osd_oi.h b/lustre/osd/osd_oi.h
index c826827..e12c4b2 100644 (file)
--- a/lustre/osd/osd_oi.h
+++ b/lustre/osd/osd_oi.h
@@ -97,13 +97,13 @@ int osd_oi_init(struct osd_thread_info *info, struct osd_oi *oi,
                 struct osd_device *osd);
 void osd_oi_fini(struct osd_thread_info *info, struct osd_oi *oi);
 
-int  osd_oi_lookup(struct osd_thread_info *info, struct osd_oi *oi,
+int  osd_oi_lookup(struct osd_thread_info *info, struct osd_device *osd,
                    const struct lu_fid *fid, struct osd_inode_id *id);
-int  osd_oi_insert(struct osd_thread_info *info, struct osd_oi *oi,
+int  osd_oi_insert(struct osd_thread_info *info, struct osd_device *osd,
                    const struct lu_fid *fid, const struct osd_inode_id *id,
                    struct thandle *th, int ingore_quota);
 int  osd_oi_delete(struct osd_thread_info *info,
-                   struct osd_oi *oi, const struct lu_fid *fid,
+                   struct osd_device *osd, const struct lu_fid *fid,
                    struct thandle *th);
 
 #endif /* __KERNEL__ */