Be sure not to take PDO lock for non-existing or remote directory.

diff --git a/lustre/ldlm/ldlm_lib.c b/lustre/ldlm/ldlm_lib.c
index f90235d..47ba3ce 100644 (file)
--- a/lustre/ldlm/ldlm_lib.c
+++ b/lustre/ldlm/ldlm_lib.c
@@ -1175,7 +1175,10 @@ static int check_for_next_transno(struct obd_device *obd)
                 CDEBUG(D_HA, "waking for next ("LPD64")\n", next_transno);
                 wake_up = 1;
         } else if (queue_len + completed == max) {
-                LASSERT(req_transno >= next_transno);
+                LASSERTF(req_transno >= next_transno,
+                         "req_transno: "LPU64", next_transno: "LPU64"\n",
+                         req_transno, next_transno);
+                        
                 CDEBUG(req_transno > obd->obd_last_committed ? D_ERROR : D_HA,
                        "waking for skipped transno (skip: "LPD64
                        ", ql: %d, comp: %d, conn: %d, next: "LPD64")\n",

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index fa75454..047d2f7 100644 (file)
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -241,19 +241,17 @@ static ldlm_mode_t mdt_lock_pdo_mode(struct mdt_thread_info *info,
 
         LASSERT(lm != LCK_MINMODE);
 
-        if (mdt_object_exists(o) > 0) {
-                /*
-                 * Ask underlaying level its opinion about possible locks.
-                 */
-                mode = mdo_lock_mode(info->mti_env, mdt_object_child(o),
-                                     mdt_dlm_mode2mdl_mode(lm));
-        } else {
-                /* 
-                 * No pdo locks possible on not existing objects, because pdo
-                 * lock is taken on parent dir and it can't absent.
-                 */
-                LBUG();
-        }
+        /* 
+         * No pdo locks possible on not existing objects, because pdo
+         * lock is taken on parent dir and it can't absent.
+         */
+        LASSERT(mdt_object_exists(o) > 0);
+
+        /*
+         * Ask underlaying level its opinion about possible locks.
+         */
+        mode = mdo_lock_mode(info->mti_env, mdt_object_child(o),
+                             mdt_dlm_mode2mdl_mode(lm));
 
         if (mode != MDL_MINMODE) {
                 /* Lower layer said what lock mode it likes to be, use it. */
@@ -1582,6 +1580,7 @@ int mdt_object_lock(struct mdt_thread_info *info, struct mdt_object *o,
         struct ldlm_namespace *ns = info->mti_mdt->mdt_namespace;
         ldlm_policy_data_t *policy = &info->mti_policy;
         struct ldlm_res_id *res_id = &info->mti_res_id;
+        int exist = mdt_object_exists(o);
         int rc;
         ENTRY;
 
@@ -1590,7 +1589,7 @@ int mdt_object_lock(struct mdt_thread_info *info, struct mdt_object *o,
         LASSERT(lh->mlh_reg_mode != LCK_MINMODE);
         LASSERT(lh->mlh_type != MDT_NUL_LOCK);
 
-        if (mdt_object_exists(o) < 0) {
+        if (exist < 0) {
                 if (locality == MDT_CROSS_LOCK) {
                         /* cross-ref object fix */
                         ibits &= ~MDS_INODELOCK_UPDATE;
@@ -1599,6 +1598,14 @@ int mdt_object_lock(struct mdt_thread_info *info, struct mdt_object *o,
                         LASSERT(!(ibits & MDS_INODELOCK_UPDATE));
                         LASSERT(ibits & MDS_INODELOCK_LOOKUP);
                 }
+                /* No PDO lock on remote object */
+                LASSERT(lh->mlh_type != MDT_PDO_LOCK);
+        } else if (exist == 0 && lh->mlh_type == MDT_PDO_LOCK) {
+                /* 
+                 * No PDO lock on non-existing object.
+                 * This may happen on removed $PWD on client.
+                 */
+                RETURN(-ESTALE);
         }
 
         memset(policy, 0, sizeof(*policy));

diff --git a/lustre/mdt/mdt_open.c b/lustre/mdt/mdt_open.c
index ce07420..f52d3a9 100644 (file)
--- a/lustre/mdt/mdt_open.c
+++ b/lustre/mdt/mdt_open.c
@@ -795,15 +795,6 @@ int mdt_reint_open(struct mdt_thread_info *info, struct mdt_lock_handle *lhc)
         if (IS_ERR(parent))
                 GOTO(out, result = PTR_ERR(parent));
 
-        result = mdt_object_exists(parent);
-        if (result == 0)
-                GOTO(out_parent, result = -ESTALE);
-        else if (result < 0) {
-                CERROR("Object "DFID" locates on remote server\n",
-                        PFID(mdt_object_fid(parent)));
-                LBUG();
-        }
-
         result = mdo_lookup(info->mti_env, mdt_object_child(parent),
                             rr->rr_name, child_fid);
         if (result != 0 && result != -ENOENT && result != -ESTALE)

diff --git a/lustre/mdt/mdt_reint.c b/lustre/mdt/mdt_reint.c
index a7b436c..aa97bab 100644 (file)
--- a/lustre/mdt/mdt_reint.c
+++ b/lustre/mdt/mdt_reint.c
@@ -61,15 +61,6 @@ static int mdt_md_create(struct mdt_thread_info *info)
         if (IS_ERR(parent))
                 RETURN(PTR_ERR(parent));
 
-        rc = mdt_object_exists(parent);
-        if (rc == 0)
-                GOTO(out, rc = -ESTALE);
-        else if (rc < 0) {
-                CERROR("Object "DFID" locates on remote server\n",
-                        PFID(mdt_object_fid(parent)));
-                LBUG();
-        }
-
         child = mdt_object_find(info->mti_env, mdt, rr->rr_fid2);
         if (!IS_ERR(child)) {
                 struct md_object *next = mdt_object_child(parent);
@@ -91,7 +82,6 @@ static int mdt_md_create(struct mdt_thread_info *info)
                 mdt_object_put(info->mti_env, child);
         } else
                 rc = PTR_ERR(child);
-out:
         mdt_object_unlock_put(info, parent, lh, rc);
         RETURN(rc);
 }
@@ -375,15 +365,6 @@ static int mdt_reint_unlink(struct mdt_thread_info *info,
         if (IS_ERR(mp))
                 GOTO(out, rc = PTR_ERR(mp));
 
-        rc = mdt_object_exists(mp);
-        if (rc == 0)
-                GOTO(out_unlock_parent, rc = -ESTALE);
-        else if (rc < 0) {
-                CERROR("Object "DFID" locates on remote server\n",
-                        PFID(mdt_object_fid(mp)));
-                LBUG();
-        }
-
         ma->ma_lmm = req_capsule_server_get(&info->mti_pill, &RMF_MDT_MD);
         ma->ma_lmm_size = req_capsule_get_size(&info->mti_pill,
                                                &RMF_MDT_MD, RCL_SERVER);