Fixes and cleanups in lmv.

[fs/lustre-release.git] / lustre / mds / mds_lmv.c

diff --git a/lustre/mds/mds_lmv.c b/lustre/mds/mds_lmv.c
index 863dba5..9dca6e3 100644 (file)
--- a/lustre/mds/mds_lmv.c
+++ b/lustre/mds/mds_lmv.c
@@ -162,7 +162,7 @@ int mds_get_lmv_attr(struct obd_device *obd, struct inode *inode,
 
        /* first calculate mea size */
         *mea_size = obd_alloc_diskmd(mds->mds_lmv_exp,
 
        /* first calculate mea size */
         *mea_size = obd_alloc_diskmd(mds->mds_lmv_exp,

-                                     (struct lov_mds_md **) mea);
+                                     (struct lov_mds_md **)mea);

        /* FIXME: error handling here */
        LASSERT(*mea != NULL);
 
        /* FIXME: error handling here */
        LASSERT(*mea != NULL);
 


@@ -411,6 +411,46 @@ int scan_and_distribute(struct obd_device *obd, struct dentry *dentry,
 
 #define MAX_DIR_SIZE    (64 * 1024)
 
 
 #define MAX_DIR_SIZE    (64 * 1024)
 

+int mds_splitting_expected(struct obd_device *obd, struct dentry *dentry)
+{
+        struct mds_obd *mds = &obd->u.mds;
+        struct mea *mea = NULL;
+        int rc, size;
+
+       /* clustered MD ? */
+       if (!mds->mds_lmv_obd)
+               RETURN(0);
+
+        /* inode exist? */
+        if (dentry->d_inode == NULL)
+                return 0;
+
+        /* a dir can be splitted only */
+        if (!S_ISDIR(dentry->d_inode->i_mode))
+                return 0;
+
+        /* large enough to be splitted? */
+        if (dentry->d_inode->i_size < MAX_DIR_SIZE)
+                return 0;
+
+        /* don't split root directory */
+        if (dentry->d_inode->i_ino == mds->mds_rootfid.id)
+                return 0;
+
+        mds_get_lmv_attr(obd, dentry->d_inode, &mea, &size);
+        if (mea) {
+                /* already splitted or slave object: shouldn't be splitted */
+                rc = 0;
+        } else {
+                /* may be splitted */
+                rc = 1;
+        }
+
+        if (mea)
+                OBD_FREE(mea, size);
+        RETURN(rc);
+}
+

 /*
  * must not be called on already splitted directories
  */
 /*
  * must not be called on already splitted directories
  */


@@ -425,22 +465,10 @@ int mds_try_to_split_dir(struct obd_device *obd,
        void *handle;
        ENTRY;
 
        void *handle;
        ENTRY;
 

-       /* clustered MD ? */
-       if (!mds->mds_lmv_obd)
-               RETURN(0);
-
-        /* don't split root directory */
-        if (dentry->d_inode->i_ino == mds->mds_rootfid.id)
-                RETURN(0);
-
-        /* we want to split only large dirs. this may be already
-         * splitted dir or a slave dir created during splitting */
-        if (dir->i_size < MAX_DIR_SIZE)
-                RETURN(0);
-
-        /* check is directory marked non-splittable */
-        if (mea && *mea)
+        /* TODO: optimization possible - we already may have mea here */
+        if (!mds_splitting_expected(obd, dentry))

                 RETURN(0);
                 RETURN(0);

+        LASSERT(mea == NULL || *mea == NULL);

 
         CDEBUG(D_OTHER, "%s: split directory %u/%lu/%lu\n",
                obd->obd_name, mds->mds_num, dir->i_ino,
 
         CDEBUG(D_OTHER, "%s: split directory %u/%lu/%lu\n",
                obd->obd_name, mds->mds_num, dir->i_ino,


@@ -459,8 +487,6 @@ int mds_try_to_split_dir(struct obd_device *obd,
                 RETURN(-ENOMEM);
         (*mea)->mea_count = nstripes;
        
                 RETURN(-ENOMEM);
         (*mea)->mea_count = nstripes;
        

-#warning "we have to take EX lock on a dir for splitting"
-        

        /* 1) create directory objects on slave MDS'es */
        /* FIXME: should this be OBD method? */
         oa = obdo_alloc();
        /* 1) create directory objects on slave MDS'es */
        /* FIXME: should this be OBD method? */
         oa = obdo_alloc();


@@ -490,7 +516,7 @@ int mds_try_to_split_dir(struct obd_device *obd,
         LASSERT(!IS_ERR(handle));
        rc = fsfilt_set_md(obd, dir, handle, *mea, mea_size);
         LASSERT(rc == 0);
         LASSERT(!IS_ERR(handle));
        rc = fsfilt_set_md(obd, dir, handle, *mea, mea_size);
         LASSERT(rc == 0);

-        fsfilt_commit(obd, dir, handle, 0);
+        fsfilt_commit(obd, mds->mds_sb, dir, handle, 0);

         LASSERT(rc == 0);
        up(&dir->i_sem);
        obdo_free(oa);
         LASSERT(rc == 0);
        up(&dir->i_sem);
        obdo_free(oa);


@@ -637,13 +663,18 @@ int mds_commitrw(int cmd, struct obd_export *exp, struct obdo *oa,
         RETURN(rc);
 }
 
         RETURN(rc);
 }
 

-int mds_choose_mdsnum(struct obd_device *obd, const char *name, int len)
+int mds_choose_mdsnum(struct obd_device *obd, const char *name, int len, int flags)

 {
 {

+        struct lmv_obd *lmv;

         struct mds_obd *mds = &obd->u.mds;
         struct mds_obd *mds = &obd->u.mds;

-        struct lmv_obd *lmv = &mds->mds_lmv_exp->exp_obd->u.lmv;
-        int i;
+        int i = mds->mds_num;

 
 

-        i = raw_name2idx(lmv->count, name, len);
+        if (flags & REC_REINT_CREATE) { 
+                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);
+        }

         RETURN(i);
 }
 
         RETURN(i);
 }