- fixes in split:

  - do not take pdo lock for non-splittable dirs if not modification is foreseen;
  - added assert to split code which checks that split should only be done in EX lock is taken.

diff --git a/lustre/cmm/cmm_object.c b/lustre/cmm/cmm_object.c
index e913acf..793e078 100644 (file)
--- a/lustre/cmm/cmm_object.c
+++ b/lustre/cmm/cmm_object.c
@@ -393,11 +393,21 @@ static mdl_mode_t cml_lock_mode(const struct lu_env *env,
                 RETURN(MDL_MINMODE);
         }
 
-        if (lm == MDL_PW && split == CMM_EXPECT_SPLIT) {
+        /*
+         * Do not take PDO lock on non-splittable objects if this is not PW,
+         * this should speed things up a bit.
+         */
+        if (split == CMM_NOT_SPLITTABLE && lm != MDL_PW)
+                RETURN(MDL_NL);
+
+        /* Protect splitting by exclusive lock. */
+        if (split == CMM_EXPECT_SPLIT && lm == MDL_PW) {
                 CDEBUG(D_INFO|D_WARNING, "Going to split "DFID"\n",
                        PFID(lu_object_fid(&mo->mo_lu)));
                 RETURN(MDL_EX);
         }
+
+        /* Have no idea about lock mode, let it be what higher layer wants. */
         RETURN(MDL_MINMODE);
 #endif
         return MDL_MINMODE;

diff --git a/lustre/cmm/cmm_split.c b/lustre/cmm/cmm_split.c
index 9192b7d..5fec095 100644 (file)
--- a/lustre/cmm/cmm_split.c
+++ b/lustre/cmm/cmm_split.c
@@ -517,6 +517,10 @@ int cmm_try_to_split(const struct lu_env *env, struct md_object *mo)
         
         if (split != CMM_EXPECT_SPLIT)
                 GOTO(cleanup, rc = 0);
+
+        LASSERTF(mo->mo_pdo_mode == MDL_EX, "Split is only valid if "
+                 "dir is protected by MDL_EX lock. Lock mode 0x%x\n",
+                 (int)mo->mo_pdo_mode);
         
         /*
          * Disable trans for splitting, since there will be so many trans in

diff --git a/lustre/include/md_object.h b/lustre/include/md_object.h
index 183096f..93318e2 100644 (file)
--- a/lustre/include/md_object.h
+++ b/lustre/include/md_object.h
@@ -298,6 +298,14 @@ struct md_object {
         struct lu_object             mo_lu;
         struct md_object_operations *mo_ops;
         struct md_dir_operations    *mo_dir_ops;
+
+        /*
+         * XXX: This is for debug only, to check on all layers lock mode on the
+         * object. This is useful say for split. Split should be protected by EX
+         * lock and this should be checked in cmm split code.
+         */
+        mdl_mode_t                   mo_reg_mode;
+        mdl_mode_t                   mo_pdo_mode;
 };
 
 static inline int lu_device_is_md(const struct lu_device *d)

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index 1d5ca70..ca13f39 100644 (file)
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -166,7 +166,7 @@ static mdl_mode_t mdt_mdl_lock_modes[] = {
         [LCK_GROUP]   = MDL_GROUP
 };
 
-static ldlm_mode_t mdt_ldlm_lock_modes[] = {
+static ldlm_mode_t mdt_dlm_lock_modes[] = {
         [MDL_MINMODE] = LCK_MINMODE,
         [MDL_EX]      = LCK_EX,
         [MDL_PW]      = LCK_PW,
@@ -177,16 +177,16 @@ static ldlm_mode_t mdt_ldlm_lock_modes[] = {
         [MDL_GROUP]   = LCK_GROUP
 };
 
-static inline mdl_mode_t mdt_ldlm_mode2mdl_mode(ldlm_mode_t mode)
+static inline mdl_mode_t mdt_dlm_mode2mdl_mode(ldlm_mode_t mode)
 {
         LASSERT(IS_PO2(mode));
         return mdt_mdl_lock_modes[mode];
 }
 
-static inline ldlm_mode_t mdt_mdl_mode2ldlm_mode(mdl_mode_t mode)
+static inline ldlm_mode_t mdt_mdl_mode2dlm_mode(mdl_mode_t mode)
 {
         LASSERT(IS_PO2(mode));
-        return mdt_ldlm_lock_modes[mode];
+        return mdt_dlm_lock_modes[mode];
 }
 
 void mdt_lock_reg_init(struct mdt_lock_handle *lh, ldlm_mode_t lm)
@@ -246,7 +246,7 @@ static ldlm_mode_t mdt_lock_pdo_mode(struct mdt_thread_info *info,
                  * Ask underlaying level its opinion about possible locks.
                  */
                 mode = mdo_lock_mode(info->mti_env, mdt_object_child(o),
-                                     mdt_ldlm_mode2mdl_mode(lm));
+                                     mdt_dlm_mode2mdl_mode(lm));
         } else {
                 /* 
                  * No pdo locks possible on not existing objects, because pdo
@@ -257,7 +257,7 @@ static ldlm_mode_t mdt_lock_pdo_mode(struct mdt_thread_info *info,
 
         if (mode != MDL_MINMODE) {
                 /* Lower layer said what lock mode it likes to be, use it. */
-                return mdt_mdl_mode2ldlm_mode(mode);
+                return mdt_mdl_mode2dlm_mode(mode);
         } else {
                 /*
                  * Lower layer does not want to specify locking mode. We od it
@@ -273,6 +273,7 @@ static ldlm_mode_t mdt_lock_pdo_mode(struct mdt_thread_info *info,
                 } else {
                         CWARN("Not expected lock type (0x%x)\n",
                               (int)mode);
+                        LBUG();
                 }
         }
 
@@ -1551,6 +1552,26 @@ struct mdt_object *mdt_object_find(const struct lu_env *env,
         RETURN(m);
 }
 
+/* XXX: This is for debug only. */
+static void mdt_object_set_mode(struct mdt_object *o,
+                                struct mdt_lock_handle *lh)
+{
+        struct md_object *n = mdt_object_child(o);
+
+        if (lh->mlh_pdo_lh.cookie != 0)
+                o->mot_obj.mo_pdo_mode = mdt_dlm_mode2mdl_mode(lh->mlh_pdo_mode);
+        else
+                o->mot_obj.mo_pdo_mode = MDL_MINMODE;
+        
+        if (lh->mlh_reg_lh.cookie != 0)
+                o->mot_obj.mo_reg_mode = mdt_dlm_mode2mdl_mode(lh->mlh_reg_mode);
+        else
+                o->mot_obj.mo_reg_mode = MDL_MINMODE;
+        
+        n->mo_pdo_mode = o->mot_obj.mo_pdo_mode;
+        n->mo_reg_mode = o->mot_obj.mo_reg_mode;
+}
+
 int mdt_object_lock(struct mdt_thread_info *info, struct mdt_object *o,
                     struct mdt_lock_handle *lh, __u64 ibits, int locality)
 {
@@ -1576,7 +1597,7 @@ int mdt_object_lock(struct mdt_thread_info *info, struct mdt_object *o,
                 }
         }
 
-        memset(policy, 0, sizeof *policy);
+        memset(policy, 0, sizeof(*policy));
         fid_build_reg_res_name(mdt_object_fid(o), res_id);
 
         /*
@@ -1585,7 +1606,8 @@ int mdt_object_lock(struct mdt_thread_info *info, struct mdt_object *o,
          */
         if (lh->mlh_type == MDT_PDO_LOCK && lh->mlh_pdo_hash != 0) {
                 lh->mlh_pdo_mode = mdt_lock_pdo_mode(info, o, lh->mlh_reg_mode);
-                if (lh->mlh_pdo_mode != LCK_MINMODE) {
+                LASSERT(lh->mlh_pdo_mode != LCK_MINMODE);
+                if (lh->mlh_pdo_mode != LCK_NL) {
                         /*
                          * Do not use LDLM_FL_LOCAL_ONLY for parallel lock, it
                          * is never going to be sent to client and we do not
@@ -1620,6 +1642,9 @@ int mdt_object_lock(struct mdt_thread_info *info, struct mdt_object *o,
                 lh->mlh_pdo_lh.cookie = 0ull;
         }
 
+        if (rc == 0)
+                mdt_object_set_mode(o, lh);
+        
         RETURN(rc);
 }
 
@@ -1651,6 +1676,9 @@ void mdt_object_unlock(struct mdt_thread_info *info, struct mdt_object *o,
                 }
                 lh->mlh_reg_lh.cookie = 0;
         }
+        
+        mdt_object_set_mode(o, lh);
+        
         EXIT;
 }