ocfs2/dlm: fix race between convert and migration

[pandora-kernel.git] / fs / ocfs2 / dlm / dlmconvert.c

diff --git a/fs/ocfs2/dlm/dlmconvert.c b/fs/ocfs2/dlm/dlmconvert.c
index 29a886d..6d97883 100644 (file)
--- a/fs/ocfs2/dlm/dlmconvert.c
+++ b/fs/ocfs2/dlm/dlmconvert.c
@@ -290,6 +290,19 @@ enum dlm_status dlmconvert_remote(struct dlm_ctxt *dlm,
                status = DLM_DENIED;
                goto bail;
        }
+
+       if (lock->ml.type == type && lock->ml.convert_type == LKM_IVMODE) {
+               mlog(0, "last convert request returned DLM_RECOVERING, but "
+                    "owner has already queued and sent ast to me. res %.*s, "
+                    "(cookie=%u:%llu, type=%d, conv=%d)\n",
+                    res->lockname.len, res->lockname.name,
+                    dlm_get_lock_cookie_node(be64_to_cpu(lock->ml.cookie)),
+                    dlm_get_lock_cookie_seq(be64_to_cpu(lock->ml.cookie)),
+                    lock->ml.type, lock->ml.convert_type);
+               status = DLM_NORMAL;
+               goto bail;
+       }
+
        res->state |= DLM_LOCK_RES_IN_PROGRESS;
        /* move lock to local convert queue */
        /* do not alter lock refcount.  switching lists. */
@@ -318,13 +331,22 @@ enum dlm_status dlmconvert_remote(struct dlm_ctxt *dlm,
 
        spin_lock(&res->spinlock);
        res->state &= ~DLM_LOCK_RES_IN_PROGRESS;
-       lock->convert_pending = 0;
-       /* if it failed, move it back to granted queue */
+       /* if it failed, move it back to granted queue.
+        * if master returns DLM_NORMAL and then down before sending ast,
+        * it may have already been moved to granted queue, reset to
+        * DLM_RECOVERING and retry convert */
        if (status != DLM_NORMAL) {
                if (status != DLM_NOTQUEUED)
                        dlm_error(status);
                dlm_revert_pending_convert(res, lock);
+       } else if (!lock->convert_pending) {
+               mlog(0, "%s: res %.*s, owner died and lock has been moved back "
+                               "to granted list, retry convert.\n",
+                               dlm->name, res->lockname.len, res->lockname.name);
+               status = DLM_RECOVERING;
        }
+
+       lock->convert_pending = 0;
 bail:
        spin_unlock(&res->spinlock);