ocfs2: wait for recovery when starting lock mastery

Signed-off-by: Kurt Hackel <kurt.hackel@oracle.com>
Signed-off-by: Mark Fasheh <mark.fasheh@oracle.com>

3 files changed, 34 insertions, 0 deletions

diff --git a/fs/ocfs2/dlm/dlmcommon.h b/fs/ocfs2/dlm/dlmcommon.h
index 829cc3948804..9bea5c6ef9bf 100644
--- a/fs/ocfs2/dlm/dlmcommon.h
+++ b/fs/ocfs2/dlm/dlmcommon.h
@@ -710,6 +710,7 @@ void dlm_wait_for_recovery(struct dlm_ctxt *dlm);
 void dlm_kick_recovery_thread(struct dlm_ctxt *dlm);
 int dlm_is_node_dead(struct dlm_ctxt *dlm, u8 node);
 int dlm_wait_for_node_death(struct dlm_ctxt *dlm, u8 node, int timeout);
+int dlm_wait_for_node_recovery(struct dlm_ctxt *dlm, u8 node, int timeout);
 
 void dlm_put(struct dlm_ctxt *dlm);
 struct dlm_ctxt *dlm_grab(struct dlm_ctxt *dlm);
diff --git a/fs/ocfs2/dlm/dlmmaster.c b/fs/ocfs2/dlm/dlmmaster.c
index 78ba77bf3a77..ed1601d1831c 100644
--- a/fs/ocfs2/dlm/dlmmaster.c
+++ b/fs/ocfs2/dlm/dlmmaster.c
@@ -899,6 +899,9 @@ redo_request:
                 } else
                         wait_on_recovery = 0;
                 spin_unlock(&dlm->spinlock);
+
+                if (wait_on_recovery)
+                        dlm_wait_for_node_recovery(dlm, bit, 10000);
         }
 
         /* must wait for lock to be mastered elsewhere */
diff --git a/fs/ocfs2/dlm/dlmrecovery.c b/fs/ocfs2/dlm/dlmrecovery.c
index c699a450b0f2..b03fecab299d 100644
--- a/fs/ocfs2/dlm/dlmrecovery.c
+++ b/fs/ocfs2/dlm/dlmrecovery.c
@@ -343,6 +343,18 @@ int dlm_is_node_dead(struct dlm_ctxt *dlm, u8 node)
         return dead;
 }
 
+/* returns true if node is no longer in the domain
+ * could be dead or just not joined */
+int dlm_is_node_recovered(struct dlm_ctxt *dlm, u8 node)
+{
+        int recovered;
+        spin_lock(&dlm->spinlock);
+        recovered = !test_bit(node, dlm->recovery_map);
+        spin_unlock(&dlm->spinlock);
+        return recovered;
+}
+
+
 int dlm_wait_for_node_death(struct dlm_ctxt *dlm, u8 node, int timeout)
 {
         if (timeout) {
@@ -361,6 +373,24 @@ int dlm_wait_for_node_death(struct dlm_ctxt *dlm, u8 node, int timeout)
         return 0;
 }
 
+int dlm_wait_for_node_recovery(struct dlm_ctxt *dlm, u8 node, int timeout)
+{
+        if (timeout) {
+                mlog(0, "%s: waiting %dms for notification of "
+                     "recovery of node %u\n", dlm->name, timeout, node);
+                wait_event_timeout(dlm->dlm_reco_thread_wq,
+                           dlm_is_node_recovered(dlm, node),
+                           msecs_to_jiffies(timeout));
+        } else {
+                mlog(0, "%s: waiting indefinitely for notification "
+                     "of recovery of node %u\n", dlm->name, node);
+                wait_event(dlm->dlm_reco_thread_wq,
+                           dlm_is_node_recovered(dlm, node));
+        }
+        /* for now, return 0 */
+        return 0;
+}
+
 /* callers of the top-level api calls (dlmlock/dlmunlock) should
  * block on the dlm->reco.event when recovery is in progress.
  * the dlm recovery thread will set this state when it begins