5 files changed, 78 insertions, 17 deletions
diff --git a/fs/ocfs2/dlm/dlmcommon.h b/fs/ocfs2/dlm/dlmcommon.h
index a106b3f2b22a..fae17c640df3 100644
--- a/fs/ocfs2/dlm/dlmcommon.h
+++ b/fs/ocfs2/dlm/dlmcommon.h
@@ -331,6 +331,7 @@ struct dlm_lock_resource
        u16 state;
        char lvb[DLM_LVB_LEN];
        unsigned int inflight_locks;
+        unsigned int inflight_assert_workers;
        unsigned long refmap[BITS_TO_LONGS(O2NM_MAX_NODES)];
 };
@@ -910,6 +911,9 @@ void dlm_lockres_drop_inflight_ref(struct dlm_ctxt *dlm,
 void dlm_lockres_grab_inflight_ref(struct dlm_ctxt *dlm,
                                   struct dlm_lock_resource *res);
+void __dlm_lockres_grab_inflight_worker(struct dlm_ctxt *dlm,
+                struct dlm_lock_resource *res);
 void dlm_queue_ast(struct dlm_ctxt *dlm, struct dlm_lock *lock);
 void dlm_queue_bast(struct dlm_ctxt *dlm, struct dlm_lock *lock);
 void __dlm_queue_ast(struct dlm_ctxt *dlm, struct dlm_lock *lock);
diff --git a/fs/ocfs2/dlm/dlmmaster.c b/fs/ocfs2/dlm/dlmmaster.c
index 3087a21d32f9..82abf0cc9a12 100644
--- a/fs/ocfs2/dlm/dlmmaster.c
+++ b/fs/ocfs2/dlm/dlmmaster.c
@@ -581,6 +581,7 @@ static void dlm_init_lockres(struct dlm_ctxt *dlm,
        atomic_set(&res->asts_reserved, 0);
        res->migration_pending = 0;
        res->inflight_locks = 0;
+        res->inflight_assert_workers = 0;
        res->dlm = dlm;
@@ -683,6 +684,43 @@ void dlm_lockres_drop_inflight_ref(struct dlm_ctxt *dlm,
        wake_up(&res->wq);
 }
+void __dlm_lockres_grab_inflight_worker(struct dlm_ctxt *dlm,
+                struct dlm_lock_resource *res)
+{
+        assert_spin_locked(&res->spinlock);
+        res->inflight_assert_workers++;
+        mlog(0, "%s:%.*s: inflight assert worker++: now %u\n",
+                        dlm->name, res->lockname.len, res->lockname.name,
+                        res->inflight_assert_workers);
+}
+static void dlm_lockres_grab_inflight_worker(struct dlm_ctxt *dlm,
+                struct dlm_lock_resource *res)
+{
+        spin_lock(&res->spinlock);
+        __dlm_lockres_grab_inflight_worker(dlm, res);
+        spin_unlock(&res->spinlock);
+}
+static void __dlm_lockres_drop_inflight_worker(struct dlm_ctxt *dlm,
+                struct dlm_lock_resource *res)
+{
+        assert_spin_locked(&res->spinlock);
+        BUG_ON(res->inflight_assert_workers == 0);
+        res->inflight_assert_workers--;
+        mlog(0, "%s:%.*s: inflight assert worker--: now %u\n",
+                        dlm->name, res->lockname.len, res->lockname.name,
+                        res->inflight_assert_workers);
+}
+static void dlm_lockres_drop_inflight_worker(struct dlm_ctxt *dlm,
+                struct dlm_lock_resource *res)
+{
+        spin_lock(&res->spinlock);
+        __dlm_lockres_drop_inflight_worker(dlm, res);
+        spin_unlock(&res->spinlock);
+}
 /*
 * lookup a lock resource by name.
 * may already exist in the hashtable.
@@ -1603,7 +1641,8 @@ send_response:
                        mlog(ML_ERROR, "failed to dispatch assert master work\n");
                        response = DLM_MASTER_RESP_ERROR;
                        dlm_lockres_put(res);
-                }
+                } else
+                        dlm_lockres_grab_inflight_worker(dlm, res);
        } else {
                if (res)
                        dlm_lockres_put(res);
@@ -2118,6 +2157,8 @@ static void dlm_assert_master_worker(struct dlm_work_item *item, void *data)
        dlm_lockres_release_ast(dlm, res);
 put:
+        dlm_lockres_drop_inflight_worker(dlm, res);
        dlm_lockres_put(res);
        mlog(0, "finished with dlm_assert_master_worker\n");
@@ -3088,11 +3129,15 @@ static int dlm_add_migration_mle(struct dlm_ctxt *dlm,
                        /* remove it so that only one mle will be found */
                        __dlm_unlink_mle(dlm, tmp);
                        __dlm_mle_detach_hb_events(dlm, tmp);
-                        ret = DLM_MIGRATE_RESPONSE_MASTERY_REF;
+                        if (tmp->type == DLM_MLE_MASTER) {
-                        mlog(0, "%s:%.*s: master=%u, newmaster=%u, "
+                                ret = DLM_MIGRATE_RESPONSE_MASTERY_REF;
-                            "telling master to get ref for cleared out mle "
+                                mlog(0, "%s:%.*s: master=%u, newmaster=%u, "
-                            "during migration\n", dlm->name, namelen, name,
+                                                "telling master to get ref "
-                            master, new_master);
+                                                "for cleared out mle during "
+                                                "migration\n", dlm->name,
+                                                namelen, name, master,
+                                                new_master);
+                        }
                }
                spin_unlock(&tmp->spinlock);
        }
diff --git a/fs/ocfs2/dlm/dlmrecovery.c b/fs/ocfs2/dlm/dlmrecovery.c
index 5de019437ea5..45067faf5695 100644
--- a/fs/ocfs2/dlm/dlmrecovery.c
+++ b/fs/ocfs2/dlm/dlmrecovery.c
@@ -1708,7 +1708,8 @@ int dlm_master_requery_handler(struct o2net_msg *msg, u32 len, void *data,
                                mlog_errno(-ENOMEM);
                                /* retry!? */
                                BUG();
-                        }
+                        } else
+                                __dlm_lockres_grab_inflight_worker(dlm, res);
                } else /* put.. incase we are not the master */
                        dlm_lockres_put(res);
                spin_unlock(&res->spinlock);
diff --git a/fs/ocfs2/dlm/dlmthread.c b/fs/ocfs2/dlm/dlmthread.c
index 9db869de829d..69aac6f088ad 100644
--- a/fs/ocfs2/dlm/dlmthread.c
+++ b/fs/ocfs2/dlm/dlmthread.c
@@ -259,12 +259,15 @@ static void dlm_run_purge_list(struct dlm_ctxt *dlm,
                 * refs on it. */
                unused = __dlm_lockres_unused(lockres);
                if (!unused ||
-                    (lockres->state & DLM_LOCK_RES_MIGRATING)) {
+                    (lockres->state & DLM_LOCK_RES_MIGRATING) ||
+                    (lockres->inflight_assert_workers != 0)) {
                        mlog(0, "%s: res %.*s is in use or being remastered, "
-                             "used %d, state %d\n", dlm->name,
+                             "used %d, state %d, assert master workers %u\n",
-                             lockres->lockname.len, lockres->lockname.name,
+                             dlm->name, lockres->lockname.len,
-                             !unused, lockres->state);
+                             lockres->lockname.name,
-                        list_move_tail(&dlm->purge_list, &lockres->purge);
+                             !unused, lockres->state,
+                             lockres->inflight_assert_workers);
+                        list_move_tail(&lockres->purge, &dlm->purge_list);
                        spin_unlock(&lockres->spinlock);
                        continue;
                }
diff --git a/fs/ocfs2/dlm/dlmunlock.c b/fs/ocfs2/dlm/dlmunlock.c
index 5698b52cf5c9..2e3c9dbab68c 100644
--- a/fs/ocfs2/dlm/dlmunlock.c
+++ b/fs/ocfs2/dlm/dlmunlock.c
@@ -191,7 +191,9 @@ static enum dlm_status dlmunlock_common(struct dlm_ctxt *dlm,
                                     DLM_UNLOCK_CLEAR_CONVERT_TYPE);
                } else if (status == DLM_RECOVERING ||
                           status == DLM_MIGRATING ||
-                           status == DLM_FORWARD) {
+                           status == DLM_FORWARD ||
+                           status == DLM_NOLOCKMGR
+                           ) {
                        /* must clear the actions because this unlock
                         * is about to be retried.  cannot free or do
                         * any list manipulation. */
@@ -200,7 +202,8 @@ static enum dlm_status dlmunlock_common(struct dlm_ctxt *dlm,
                             res->lockname.name,
                             status==DLM_RECOVERING?"recovering":
                             (status==DLM_MIGRATING?"migrating":
-                              "forward"));
+                                (status == DLM_FORWARD ? "forward" :
+                                                "nolockmanager")));
                        actions = 0;
                }
                if (flags & LKM_CANCEL)
@@ -364,7 +367,10 @@ static enum dlm_status dlm_send_remote_unlock_request(struct dlm_ctxt *dlm,
                         * updated state to the recovery master.  this thread
                         * just needs to finish out the operation and call
                         * the unlockast. */
-                        ret = DLM_NORMAL;
+                        if (dlm_is_node_dead(dlm, owner))
+                                ret = DLM_NORMAL;
+                        else
+                                ret = DLM_NOLOCKMGR;
                } else {
                        /* something bad.  this will BUG in ocfs2 */
                        ret = dlm_err_to_dlm_status(tmpret);
@@ -638,7 +644,9 @@ retry:
        if (status == DLM_RECOVERING ||
            status == DLM_MIGRATING ||
-            status == DLM_FORWARD) {
+            status == DLM_FORWARD ||
+            status == DLM_NOLOCKMGR) {
                /* We want to go away for a tiny bit to allow recovery
                 * / migration to complete on this resource. I don't
                 * know of any wait queue we could sleep on as this
@@ -650,7 +658,7 @@ retry:
                msleep(50);
                mlog(0, "retrying unlock due to pending recovery/"
-                     "migration/in-progress\n");
+                     "migration/in-progress/reconnect\n");
                goto retry;
        }