Add flock support.

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

diff --git a/lustre/mds/mds_lmv.c b/lustre/mds/mds_lmv.c
index 0b7eeac..8d5a431 100644 (file)
--- a/lustre/mds/mds_lmv.c
+++ b/lustre/mds/mds_lmv.c
@@ -57,20 +57,26 @@ int mds_lmv_connect(struct obd_device *obd, char * lmv_name)
         if (mds->mds_lmv_obd)
                 RETURN(0);
 
+        down(&mds->mds_lmv_sem);
+        if (mds->mds_lmv_obd) {
+                up(&mds->mds_lmv_sem);
+                RETURN(0);
+        }
+
         mds->mds_lmv_obd = class_name2obd(lmv_name);
         if (!mds->mds_lmv_obd) {
                 CERROR("MDS cannot locate LMV %s\n",
                        lmv_name);
                 mds->mds_lmv_obd = ERR_PTR(-ENOTCONN);
-                RETURN(-ENOTCONN);
+                GOTO(err_last, rc = -ENOTCONN);
         }
 
-        rc = obd_connect(&conn, mds->mds_lmv_obd, &obd->obd_uuid);
+        rc = obd_connect(&conn, mds->mds_lmv_obd, &obd->obd_uuid, OBD_OPT_MDS_CONNECTION);
         if (rc) {
                 CERROR("MDS cannot connect to LMV %s (%d)\n",
                        lmv_name, rc);
                 mds->mds_lmv_obd = ERR_PTR(rc);
-                RETURN(rc);
+                GOTO(err_last, rc);
         }
         mds->mds_lmv_exp = class_conn2export(&conn);
         if (mds->mds_lmv_exp == NULL)
@@ -104,7 +110,8 @@ int mds_lmv_connect(struct obd_device *obd, char * lmv_name)
                           "inter_mds", 0, NULL);
         if (rc)
                 GOTO(err_reg, rc);
-        
+
+        up(&mds->mds_lmv_sem);
        RETURN(0);
 
 err_reg:
@@ -113,6 +120,8 @@ err_discon:
         obd_disconnect(mds->mds_lmv_exp, 0);
         mds->mds_lmv_exp = NULL;
         mds->mds_lmv_obd = ERR_PTR(rc);
+err_last:
+        up(&mds->mds_lmv_sem);
         RETURN(rc);
 }
 
@@ -135,6 +144,7 @@ int mds_lmv_disconnect(struct obd_device *obd, int flags)
         int rc = 0;
         ENTRY;
 
+        down(&mds->mds_lmv_sem);
         if (!IS_ERR(mds->mds_lmv_obd) && mds->mds_lmv_exp != NULL) {
                 obd_register_observer(mds->mds_lmv_obd, NULL);
 
@@ -148,6 +158,7 @@ int mds_lmv_disconnect(struct obd_device *obd, int flags)
                 mds->mds_lmv_exp = NULL;
                 mds->mds_lmv_obd = NULL;
         }
+        up(&mds->mds_lmv_sem);
 
         RETURN(rc);
 }
@@ -161,6 +172,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,16 +181,14 @@ 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);
                *mea = NULL;
        } else
                 rc = 0;
-                        
+
        RETURN(rc);
 }
 
@@ -477,6 +488,8 @@ cleanup:
 
 #define MAX_DIR_SIZE    (64 * 1024)
 
+#define I_NON_SPLITTABLE        256
+
 int mds_splitting_expected(struct obd_device *obd, struct dentry *dentry)
 {
         struct mds_obd *mds = &obd->u.mds;
@@ -495,6 +508,10 @@ int mds_splitting_expected(struct obd_device *obd, struct dentry *dentry)
         if (!S_ISDIR(dentry->d_inode->i_mode))
                 return MDS_NO_SPLITTABLE;
 
+        /* already splittied or slave directory (part of splitted dir) */
+        if (dentry->d_inode->i_flags & I_NON_SPLITTABLE)
+                return MDS_NO_SPLITTABLE;
+
         /* don't split root directory */
         if (dentry->d_inode->i_ino == mds->mds_rootfid.id)
                 return MDS_NO_SPLITTABLE;
@@ -507,6 +524,8 @@ int mds_splitting_expected(struct obd_device *obd, struct dentry *dentry)
         if (mea) {
                 /* already splitted or slave object: shouldn't be splitted */
                 rc = MDS_NO_SPLITTABLE;
+                /* mark to skip subsequent checks */
+                dentry->d_inode->i_flags |= I_NON_SPLITTABLE;
         } else {
                 /* may be splitted */
                 rc = MDS_EXPECT_SPLIT;
@@ -521,7 +540,7 @@ int mds_splitting_expected(struct obd_device *obd, struct dentry *dentry)
  * must not be called on already splitted directories.
  */
 int mds_try_to_split_dir(struct obd_device *obd, struct dentry *dentry,
-                         struct mea **mea, int nstripes)
+                         struct mea **mea, int nstripes, int update_mode)
 {
         struct inode *dir = dentry->d_inode;
         struct mds_obd *mds = &obd->u.mds;
@@ -531,6 +550,8 @@ int mds_try_to_split_dir(struct obd_device *obd, struct dentry *dentry,
        void *handle;
        ENTRY;
 
+        if (update_mode != LCK_EX)
+                return 0;
         /* TODO: optimization possible - we already may have mea here */
         rc = mds_splitting_expected(obd, dentry);
         if (rc == MDS_NO_SPLITTABLE)
@@ -791,7 +812,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);
 }
@@ -1041,3 +1062,62 @@ cleanup:
         RETURN(rc);
 }
 
+int mds_convert_mea_ea(struct obd_device *obd, struct inode *inode,
+                       struct lov_mds_md *lmm, int lmmsize)
+{
+        int i, rc, err, size;
+        struct mea_old *old;
+        struct mea *mea;
+        struct mea *new;
+        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 || lmmsize < 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);
+
+        size = sizeof(struct ll_fid) * old->mea_count + sizeof(struct mea);
+        OBD_ALLOC(new, 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);
+        }
+
+        rc = fsfilt_set_md(obd, inode, handle, (struct lov_mds_md *) new, size);
+        if (rc > lmmsize)
+                size = lmmsize;
+        memcpy(lmm, new, size);
+
+        err = fsfilt_commit(obd, obd->u.mds.mds_sb, inode, handle, 0);
+        if (!rc)
+                rc = err ? err : size;
+        GOTO(conv_free, rc);
+conv_free:
+        OBD_FREE(new, size);
+        return rc;
+}
+