b=3983

- adds a bit more balanced hash that takes all chars into account
- stores hash type in mea
- converts old non-marked mea to new one with magic

diff --git a/lustre/include/linux/lustre_idl.h b/lustre/include/linux/lustre_idl.h
index 14cdee1..db2a810 100644 (file)
--- a/lustre/include/linux/lustre_idl.h
+++ b/lustre/include/linux/lustre_idl.h
@@ -518,7 +518,17 @@ struct ll_fid {
         __u32 padding;
 };
 
+struct mea_old {
+        __u32 mea_count;
+        __u32 mea_master;
+        struct ll_fid mea_fids[0];
+};
+
+#define MEA_MAGIC_LAST_CHAR      0xb2221ca1
+#define MEA_MAGIC_ALL_CHARS      0xb222a11c
+
 struct mea {
+        __u32 mea_magic;
         __u32 mea_count;
         __u32 mea_master;
         struct ll_fid mea_fids[0];

diff --git a/lustre/include/linux/obd_class.h b/lustre/include/linux/obd_class.h
index d7e847d..c371aff 100644 (file)
--- a/lustre/include/linux/obd_class.h
+++ b/lustre/include/linux/obd_class.h
@@ -1506,6 +1506,6 @@ void class_exit_uuidlist(void);
 
 /* mea.c */
 int mea_name2idx(struct mea *mea, char *name, int namelen);
-int raw_name2idx(int count, const char *name, int namelen);
+int raw_name2idx(int hashtype, int count, const char *name, int namelen);
 
 #endif /* __LINUX_OBD_CLASS_H */

diff --git a/lustre/lmv/lmv_intent.c b/lustre/lmv/lmv_intent.c
index 1e0e800..925e55d 100644 (file)
--- a/lustre/lmv/lmv_intent.c
+++ b/lustre/lmv/lmv_intent.c
@@ -143,7 +143,7 @@ repeat:
         if (obj) {
                 /* directory is already splitted, so we have to forward
                  * request to the right MDS */
-                mds = raw_name2idx(obj->objcount, (char *)name, len);
+                mds = raw_name2idx(obj->hashtype, obj->objcount, (char *)name, len);
                 CDEBUG(D_OTHER, "forward to MDS #%u (%lu/%lu/%lu)\n", mds,
                        (unsigned long) rpfid.mds, (unsigned long) rpfid.id,
                        (unsigned long) rpfid.generation);
@@ -280,7 +280,7 @@ int lmv_intent_getattr(struct obd_export *exp, struct ll_uctxt *uctxt,
         obj = lmv_grab_obj(obd, pfid);
         if (obj && len) {
                 /* directory is already splitted. calculate mds */
-                mds = raw_name2idx(obj->objcount, (char *) name, len);
+                mds = raw_name2idx(obj->hashtype, obj->objcount, (char *) name, len);
                 rpfid = obj->objs[mds].fid;
                 mds = rpfid.mds;
                 lmv_put_obj(obj);
@@ -492,7 +492,7 @@ int lmv_intent_lookup(struct obd_export *exp, struct ll_uctxt *uctxt,
                 rpfid = *pfid;
                 obj = lmv_grab_obj(obd, pfid);
                 if (obj) {
-                        mds = raw_name2idx(obj->objcount, (char *) name, len);
+                        mds = raw_name2idx(obj->hashtype, obj->objcount, (char *) name, len);
                         rpfid = obj->objs[mds].fid;
                         lmv_put_obj(obj);
                 }
@@ -518,7 +518,7 @@ repeat:
         if (obj) {
                 if (len) {
                         /* directory is already splitted. calculate mds */
-                        mds = raw_name2idx(obj->objcount, (char *)name, len);
+                        mds = raw_name2idx(obj->hashtype, obj->objcount, (char *)name, len);
                         rpfid = obj->objs[mds].fid;
                         mds = rpfid.mds;
                 }

diff --git a/lustre/lmv/lmv_internal.h b/lustre/lmv/lmv_internal.h
index 377fb43..2876962 100644 (file)
--- a/lustre/lmv/lmv_internal.h
+++ b/lustre/lmv/lmv_internal.h
@@ -23,6 +23,7 @@ struct lmv_obj {
         atomic_t           count;          /* ref counter. */
         struct ll_fid      fid;            /* master fid of dir */
         void               *update;        /* bitmap of status (uptodate) */
+       __u32              hashtype;
         int                objcount;       /* number of slaves */
         struct lmv_inode   *objs;          /* array of dirobjs */
         struct obd_device  *obd;           /* pointer to LMV itself */

diff --git a/lustre/lmv/lmv_obd.c b/lustre/lmv/lmv_obd.c
index aa24b74..dadea41 100644 (file)
--- a/lustre/lmv/lmv_obd.c
+++ b/lustre/lmv/lmv_obd.c
@@ -659,7 +659,7 @@ static int lmv_change_cbdata_name(struct obd_export *exp, struct ll_fid *pfid,
         obj = lmv_grab_obj(obd, pfid);
         if (obj) {
                 /* directory is splitted. look for right mds for this name. */
-                mds = raw_name2idx(obj->objcount, name, len);
+                mds = raw_name2idx(obj->hashtype, obj->objcount, name, len);
                 mds = obj->objs[mds].fid.mds;
                 lmv_put_obj(obj);
         }
@@ -767,7 +767,7 @@ repeat:
         LASSERT(++loop <= 2);
         obj = lmv_grab_obj(obd, &op_data->fid1);
         if (obj) {
-                mds = raw_name2idx(obj->objcount, op_data->name,
+                mds = raw_name2idx(obj->hashtype, obj->objcount, op_data->name,
                                    op_data->namelen);
                 op_data->fid1 = obj->objs[mds].fid;
                 lmv_put_obj(obj);
@@ -904,8 +904,8 @@ int lmv_enqueue(struct obd_export *exp, int lock_type,
                 if (obj) {
                         /* directory is splitted. look for right mds for this
                          * name */
-                        mds = raw_name2idx(obj->objcount, (char *)data->name,
-                                           data->namelen);
+                        mds = raw_name2idx(obj->hashtype, obj->objcount,
+                                           (char *)data->name, data->namelen);
                         data->fid1 = obj->objs[mds].fid;
                         lmv_put_obj(obj);
                 }
@@ -939,7 +939,7 @@ repeat:
         obj = lmv_grab_obj(obd, fid);
         if (obj) {
                 /* directory is splitted. look for right mds for this name */
-                mds = raw_name2idx(obj->objcount, filename, namelen - 1);
+                mds = raw_name2idx(obj->hashtype, obj->objcount, filename, namelen - 1);
                 rfid = obj->objs[mds].fid;
                 lmv_put_obj(obj);
         }
@@ -1004,7 +1004,7 @@ int lmv_link(struct obd_export *exp, struct mdc_op_data *data,
                 /* usual link request */
                 obj = lmv_grab_obj(obd, &data->fid1);
                 if (obj) {
-                        rc = raw_name2idx(obj->objcount, data->name,
+                        rc = raw_name2idx(obj->hashtype, obj->objcount, data->name,
                                           data->namelen);
                         data->fid1 = obj->objs[rc].fid;
                         lmv_put_obj(obj);
@@ -1082,7 +1082,7 @@ int lmv_rename(struct obd_export *exp, struct mdc_op_data *data,
         if (obj) {
                 /* directory is already splitted, so we have to forward request
                  * to the right MDS */
-                mds = raw_name2idx(obj->objcount, (char *)old, oldlen);
+                mds = raw_name2idx(obj->hashtype, obj->objcount, (char *)old, oldlen);
                 data->fid1 = obj->objs[mds].fid;
                 CDEBUG(D_OTHER, "forward to MDS #%u (%lu/%lu/%lu)\n", mds,
                        (unsigned long)obj->objs[mds].fid.mds,
@@ -1095,7 +1095,7 @@ int lmv_rename(struct obd_export *exp, struct mdc_op_data *data,
         if (obj) {
                 /* directory is already splitted, so we have to forward request
                  * to the right MDS */
-                mds = raw_name2idx(obj->objcount, (char *)new, newlen);
+                mds = raw_name2idx(obj->hashtype, obj->objcount, (char *)new, newlen);
                 data->fid2 = obj->objs[mds].fid;
                 CDEBUG(D_OTHER, "forward to MDS #%u (%lu/%lu/%lu)\n", mds,
                        (unsigned long)obj->objs[mds].fid.mds,
@@ -1358,7 +1358,7 @@ int lmv_unlink(struct obd_export *exp, struct mdc_op_data *data,
                 
                 obj = lmv_grab_obj(obd, &data->fid1);
                 if (obj) {
-                        i = raw_name2idx(obj->objcount, data->name,
+                        i = raw_name2idx(obj->hashtype, obj->objcount, data->name,
                                          data->namelen);
                         data->fid1 = obj->objs[i].fid;
                         lmv_put_obj(obj);
@@ -1498,6 +1498,7 @@ int lmv_obd_create(struct obd_export *exp, struct obdo *oa,
         mea = (struct mea *)*ea;
         if (!mea->mea_count || mea->mea_count > lmv->desc.ld_tgt_count)
                 mea->mea_count = lmv->desc.ld_tgt_count;
+        mea->mea_magic = MEA_MAGIC_ALL_CHARS;
 
         mea->mea_master = -1;
         lcount = lmv->desc.ld_tgt_count;

diff --git a/lustre/lmv/lmv_objmgr.c b/lustre/lmv/lmv_objmgr.c
index fd9eeef..9485f9d 100644 (file)
--- a/lustre/lmv/lmv_objmgr.c
+++ b/lustre/lmv/lmv_objmgr.c
@@ -61,6 +61,9 @@ lmv_alloc_obj(struct obd_device *obd, struct ll_fid *fid,
         unsigned int obj_size;
         struct lmv_obd *lmv = &obd->u.lmv;
 
+        LASSERT(mea->mea_magic == MEA_MAGIC_LAST_CHAR
+                        || mea->mea_magic == MEA_MAGIC_ALL_CHARS);
+
         OBD_ALLOC(obj, sizeof(*obj));
         if (!obj)
                 return NULL;
@@ -68,6 +71,7 @@ lmv_alloc_obj(struct obd_device *obd, struct ll_fid *fid,
         obj->obd = obd;
         obj->state = 0;
         obj->fid = *fid;
+        obj->hashtype = mea->mea_magic;
           
         init_MUTEX(&obj->guard);
         atomic_set(&obj->count, 0);

diff --git a/lustre/mds/handler.c b/lustre/mds/handler.c
index 4dfa0b8..535b578 100644 (file)
--- a/lustre/mds/handler.c
+++ b/lustre/mds/handler.c
@@ -598,6 +598,8 @@ int mds_get_md(struct obd_device *obd, struct inode *inode, void *md,
                 
                 if (S_ISREG(inode->i_mode))
                         rc = mds_convert_lov_ea(obd, inode, md, lmm_size);
+                if (S_ISDIR(inode->i_mode))
+                        rc = mds_convert_mea_ea(obd, inode, md, lmm_size);
 
                 if (rc == 0) {
                         *size = lmm_size;
@@ -1465,6 +1467,8 @@ repeat:
                 OBD_ALLOC(mea, mealen);
                 if (mea == NULL)
                         GOTO(cleanup, rc = -ENOMEM);
+                mea->mea_magic = MEA_MAGIC_ALL_CHARS;
+                mea->mea_master = 0;
                 mea->mea_count = 0;
                 down(&new->d_inode->i_sem);
                 rc = fsfilt_set_md(obd, new->d_inode, handle, mea, mealen);

diff --git a/lustre/mds/mds_internal.h b/lustre/mds/mds_internal.h
index 8adf159..8fed0cb 100644 (file)
--- a/lustre/mds/mds_internal.h
+++ b/lustre/mds/mds_internal.h
@@ -146,6 +146,7 @@ int mds_unlink_slave_objs(struct obd_device *, struct dentry *);
 void mds_unlock_slave_objs(struct obd_device *, struct dentry *,
                            struct lustre_handle *);
 int mds_lock_and_check_slave(int, struct ptlrpc_request *, struct lustre_handle *);
+int mds_convert_mea_ea(struct obd_device *, struct inode *, struct lov_mds_md *, int);
 
 
 #endif /* _MDS_INTERNAL_H */

diff --git a/lustre/mds/mds_lmv.c b/lustre/mds/mds_lmv.c
index 6e97e63..55d10be 100644 (file)
--- a/lustre/mds/mds_lmv.c
+++ b/lustre/mds/mds_lmv.c
@@ -161,6 +161,8 @@ int mds_get_lmv_attr(struct obd_device *obd, struct inode *inode,
 
        if (!mds->mds_lmv_obd)
                RETURN(0);
+        if (!S_ISDIR(inode->i_mode))
+                RETURN(0);
 
        /* first calculate mea size */
         *mea_size = obd_alloc_diskmd(mds->mds_lmv_exp,
@@ -168,9 +170,7 @@ int mds_get_lmv_attr(struct obd_device *obd, struct inode *inode,
         if (*mea_size < 0 || *mea == NULL)
                 return *mea_size < 0 ? *mea_size : -EINVAL;
 
-       down(&inode->i_sem);
-       rc = fsfilt_get_md(obd, inode, *mea, *mea_size);
-       up(&inode->i_sem);
+        rc = mds_get_md(obd, inode, *mea, mea_size, 1);
 
        if (rc <= 0) {
                OBD_FREE(*mea, *mea_size);
@@ -801,7 +801,7 @@ int mds_choose_mdsnum(struct obd_device *obd, const char *name, int len, int fla
                 i = mds->mds_num;
         } else if (mds->mds_lmv_exp) {
                 lmv = &mds->mds_lmv_exp->exp_obd->u.lmv;
-                i = raw_name2idx(lmv->desc.ld_tgt_count, name, len);
+                i = raw_name2idx(MEA_MAGIC_LAST_CHAR, lmv->desc.ld_tgt_count, name, len);
         }
         RETURN(i);
 }
@@ -1051,3 +1051,60 @@ cleanup:
         RETURN(rc);
 }
 
+int mds_convert_mea_ea(struct obd_device *obd, struct inode *inode,
+                       struct lov_mds_md *lmm, int lmm_size)
+{
+        struct mea_old *old;
+        struct mea *mea;
+        struct mea *new;
+        int rc, err, i;
+        void *handle;
+        ENTRY;
+
+        mea = (struct mea *) lmm;
+        if (mea->mea_magic == MEA_MAGIC_LAST_CHAR ||
+                mea->mea_magic == MEA_MAGIC_ALL_CHARS)
+                RETURN(0);
+
+        old = (struct mea_old *) lmm;
+        rc = sizeof(struct ll_fid) * old->mea_count + sizeof(struct mea_old);
+        if (old->mea_count > 256 || old->mea_master > 256 || lmm_size < rc
+                        || old->mea_master > old->mea_count) {
+                CWARN("unknown MEA format, dont convert it\n");
+                CWARN("  count %u, master %u, size %u\n",
+                      old->mea_count, old->mea_master, rc);
+                RETURN(0);
+        }
+                
+        CWARN("converting MEA EA on %lu/%u from V0 to V1 (%u/%u)\n",
+              inode->i_ino, inode->i_generation, old->mea_count, old->mea_master);
+
+        lmm_size = sizeof(struct ll_fid) * old->mea_count + sizeof(struct mea);
+        OBD_ALLOC(new, lmm_size);
+        if (new == NULL)
+                RETURN(-ENOMEM);
+
+        new->mea_magic = MEA_MAGIC_LAST_CHAR;
+        new->mea_count = old->mea_count;
+        new->mea_master = old->mea_master;
+        for (i = 0; i < new->mea_count; i++)
+                new->mea_fids[i] = old->mea_fids[i];
+
+        handle = fsfilt_start(obd, inode, FSFILT_OP_SETATTR, NULL);
+        if (IS_ERR(handle)) {
+                rc = PTR_ERR(handle);
+                GOTO(conv_free, rc);
+        }
+
+        lmm = (struct lov_mds_md *) new;
+        rc = fsfilt_set_md(obd, inode, handle, lmm, lmm_size);
+
+        err = fsfilt_commit(obd, obd->u.mds.mds_sb, inode, handle, 0);
+        if (!rc)
+                rc = err ? err : lmm_size;
+        GOTO(conv_free, rc);
+conv_free:
+        OBD_FREE(new, lmm_size);
+        return rc;
+}
+

diff --git a/lustre/obdclass/mea.c b/lustre/obdclass/mea.c
index fa9ce2a..ec276fe 100644 (file)
--- a/lustre/obdclass/mea.c
+++ b/lustre/obdclass/mea.c
@@ -34,43 +34,63 @@
 #endif
 #include <linux/lprocfs_status.h>
 
-
-int mea_name2idx(struct mea *mea, char *name, int namelen)
+static int mea_last_char_hash(int count, char *name, int namelen)
 {
         unsigned int c;
-
-       /* just to simplify caller code */
-               if (mea == NULL)
-               return 0;
-
-        if (mea->mea_count == 0)
-                return 0;
-
-        /* FIXME: real hash calculation here */
+        
         c = name[namelen - 1];
         if (c == 0)
                 CWARN("looks like wrong len is passed\n");
-        c = c % mea->mea_count;
-       
-       LASSERT(c < mea->mea_count);
+        c = c % count;
         return c;
 }
 
-int raw_name2idx(int count, const char *name, int namelen)
+static int mea_all_chars_hash(int count, char *name, int namelen)
 {
-        unsigned int c;
+        unsigned int c = 0;
+
+        while (--namelen >= 0)
+                c += name[namelen];
+        c = c % count;
+        return c;
+}
+
+int raw_name2idx(int hashtype, int count, const char *name, int namelen)
+{
+        unsigned int c = 0;
 
         LASSERT(namelen > 0);
         if (count <= 1)
                 return 0;
 
-
-        /* FIXME: real hash calculation here */
-        c = name[namelen - 1];
-        if (c == 0)
-                CWARN("looks like wrong len is passed\n");
-        c = c % count;
+        switch (hashtype) {
+                case MEA_MAGIC_LAST_CHAR:
+                        c = mea_last_char_hash(count, (char *) name, namelen);
+                        break;
+                case MEA_MAGIC_ALL_CHARS:
+                        c = mea_all_chars_hash(count, (char *) name, namelen);
+                        break;
+                default:
+                        CERROR("unknown hash type 0x%x\n", hashtype);
+        }
        
         return c;
 }
 
+int mea_name2idx(struct mea *mea, char *name, int namelen)
+{
+        unsigned int c;
+
+       /* just to simplify caller code */
+               if (mea == NULL)
+               return 0;
+
+        if (mea->mea_count == 0)
+                return 0;
+
+       c = raw_name2idx(mea->mea_magic, mea->mea_count, name, namelen);
+       LASSERT(c < mea->mea_count);
+        return c;
+}
+
+