1 files changed, 82 insertions, 82 deletions

diff --git a/fs/ocfs2/dlm/dlmrecovery.c b/fs/ocfs2/dlm/dlmrecovery.c
index 7efab6d28a21..01ebfd0bdad7 100644
--- a/fs/ocfs2/dlm/dlmrecovery.c
+++ b/fs/ocfs2/dlm/dlmrecovery.c
@@ -362,40 +362,38 @@ static int dlm_is_node_recovered(struct dlm_ctxt *dlm, u8 node)
 }
 
 
-int dlm_wait_for_node_death(struct dlm_ctxt *dlm, u8 node, int timeout)
+void dlm_wait_for_node_death(struct dlm_ctxt *dlm, u8 node, int timeout)
 {
-        if (timeout) {
-                mlog(ML_NOTICE, "%s: waiting %dms for notification of "
-                     "death of node %u\n", dlm->name, timeout, node);
+        if (dlm_is_node_dead(dlm, node))
+                return;
+
+        printk(KERN_NOTICE "o2dlm: Waiting on the death of node %u in "
+               "domain %s\n", node, dlm->name);
+
+        if (timeout)
                 wait_event_timeout(dlm->dlm_reco_thread_wq,
-                           dlm_is_node_dead(dlm, node),
-                           msecs_to_jiffies(timeout));
-        } else {
-                mlog(ML_NOTICE, "%s: waiting indefinitely for notification "
-                     "of death of node %u\n", dlm->name, node);
+                                   dlm_is_node_dead(dlm, node),
+                                   msecs_to_jiffies(timeout));
+        else
                 wait_event(dlm->dlm_reco_thread_wq,
                            dlm_is_node_dead(dlm, node));
-        }
-        /* for now, return 0 */
-        return 0;
 }
 
-int dlm_wait_for_node_recovery(struct dlm_ctxt *dlm, u8 node, int timeout)
+void 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);
+        if (dlm_is_node_recovered(dlm, node))
+                return;
+
+        printk(KERN_NOTICE "o2dlm: Waiting on the recovery of node %u in "
+               "domain %s\n", node, dlm->name);
+
+        if (timeout)
                 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);
+                                   dlm_is_node_recovered(dlm, node),
+                                   msecs_to_jiffies(timeout));
+        else
                 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
@@ -430,6 +428,8 @@ static void dlm_begin_recovery(struct dlm_ctxt *dlm)
 {
         spin_lock(&dlm->spinlock);
         BUG_ON(dlm->reco.state & DLM_RECO_STATE_ACTIVE);
+        printk(KERN_NOTICE "o2dlm: Begin recovery on domain %s for node %u\n",
+               dlm->name, dlm->reco.dead_node);
         dlm->reco.state |= DLM_RECO_STATE_ACTIVE;
         spin_unlock(&dlm->spinlock);
 }
@@ -440,9 +440,18 @@ static void dlm_end_recovery(struct dlm_ctxt *dlm)
         BUG_ON(!(dlm->reco.state & DLM_RECO_STATE_ACTIVE));
         dlm->reco.state &= ~DLM_RECO_STATE_ACTIVE;
         spin_unlock(&dlm->spinlock);
+        printk(KERN_NOTICE "o2dlm: End recovery on domain %s\n", dlm->name);
         wake_up(&dlm->reco.event);
 }
 
+static void dlm_print_recovery_master(struct dlm_ctxt *dlm)
+{
+        printk(KERN_NOTICE "o2dlm: Node %u (%s) is the Recovery Master for the "
+               "dead node %u in domain %s\n", dlm->reco.new_master,
+               (dlm->node_num == dlm->reco.new_master ? "me" : "he"),
+               dlm->reco.dead_node, dlm->name);
+}
+
 static int dlm_do_recovery(struct dlm_ctxt *dlm)
 {
         int status = 0;
@@ -505,9 +514,8 @@ static int dlm_do_recovery(struct dlm_ctxt *dlm)
                 }
                 mlog(0, "another node will master this recovery session.\n");
         }
-        mlog(0, "dlm=%s (%d), new_master=%u, this node=%u, dead_node=%u\n",
-             dlm->name, task_pid_nr(dlm->dlm_reco_thread_task), dlm->reco.new_master,
-             dlm->node_num, dlm->reco.dead_node);
+
+        dlm_print_recovery_master(dlm);
 
         /* it is safe to start everything back up here
          * because all of the dead node's lock resources
@@ -518,15 +526,13 @@ static int dlm_do_recovery(struct dlm_ctxt *dlm)
         return 0;
 
 master_here:
-        mlog(ML_NOTICE, "(%d) Node %u is the Recovery Master for the Dead Node "
-             "%u for Domain %s\n", task_pid_nr(dlm->dlm_reco_thread_task),
-             dlm->node_num, dlm->reco.dead_node, dlm->name);
+        dlm_print_recovery_master(dlm);
 
         status = dlm_remaster_locks(dlm, dlm->reco.dead_node);
         if (status < 0) {
                 /* we should never hit this anymore */
-                mlog(ML_ERROR, "error %d remastering locks for node %u, "
-                     "retrying.\n", status, dlm->reco.dead_node);
+                mlog(ML_ERROR, "%s: Error %d remastering locks for node %u, "
+                     "retrying.\n", dlm->name, status, dlm->reco.dead_node);
                 /* yield a bit to allow any final network messages
                  * to get handled on remaining nodes */
                 msleep(100);
@@ -567,7 +573,7 @@ static int dlm_remaster_locks(struct dlm_ctxt *dlm, u8 dead_node)
                 BUG_ON(ndata->state != DLM_RECO_NODE_DATA_INIT);
                 ndata->state = DLM_RECO_NODE_DATA_REQUESTING;
 
-                mlog(0, "requesting lock info from node %u\n",
+                mlog(0, "%s: Requesting lock info from node %u\n", dlm->name,
                      ndata->node_num);
 
                 if (ndata->node_num == dlm->node_num) {
@@ -640,7 +646,7 @@ static int dlm_remaster_locks(struct dlm_ctxt *dlm, u8 dead_node)
                 spin_unlock(&dlm_reco_state_lock);
         }
 
-        mlog(0, "done requesting all lock info\n");
+        mlog(0, "%s: Done requesting all lock info\n", dlm->name);
 
         /* nodes should be sending reco data now
          * just need to wait */
@@ -802,10 +808,9 @@ static int dlm_request_all_locks(struct dlm_ctxt *dlm, u8 request_from,
 
         /* negative status is handled by caller */
         if (ret < 0)
-                mlog(ML_ERROR, "Error %d when sending message %u (key "
-                     "0x%x) to node %u\n", ret, DLM_LOCK_REQUEST_MSG,
-                     dlm->key, request_from);
-
+                mlog(ML_ERROR, "%s: Error %d send LOCK_REQUEST to node %u "
+                     "to recover dead node %u\n", dlm->name, ret,
+                     request_from, dead_node);
         // return from here, then
         // sleep until all received or error
         return ret;
@@ -956,9 +961,9 @@ static int dlm_send_all_done_msg(struct dlm_ctxt *dlm, u8 dead_node, u8 send_to)
         ret = o2net_send_message(DLM_RECO_DATA_DONE_MSG, dlm->key, &done_msg,
                                  sizeof(done_msg), send_to, &tmpret);
         if (ret < 0) {
-                mlog(ML_ERROR, "Error %d when sending message %u (key "
-                     "0x%x) to node %u\n", ret, DLM_RECO_DATA_DONE_MSG,
-                     dlm->key, send_to);
+                mlog(ML_ERROR, "%s: Error %d send RECO_DATA_DONE to node %u "
+                     "to recover dead node %u\n", dlm->name, ret, send_to,
+                     dead_node);
                 if (!dlm_is_host_down(ret)) {
                         BUG();
                 }
@@ -1127,9 +1132,11 @@ static int dlm_send_mig_lockres_msg(struct dlm_ctxt *dlm,
         if (ret < 0) {
                 /* XXX: negative status is not handled.
                  * this will end up killing this node. */
-                mlog(ML_ERROR, "Error %d when sending message %u (key "
-                     "0x%x) to node %u\n", ret, DLM_MIG_LOCKRES_MSG,
-                     dlm->key, send_to);
+                mlog(ML_ERROR, "%s: res %.*s, Error %d send MIG_LOCKRES to "
+                     "node %u (%s)\n", dlm->name, mres->lockname_len,
+                     mres->lockname, ret, send_to,
+                     (orig_flags & DLM_MRES_MIGRATION ?
+                      "migration" : "recovery"));
         } else {
                 /* might get an -ENOMEM back here */
                 ret = status;
@@ -1767,7 +1774,7 @@ static int dlm_process_recovery_data(struct dlm_ctxt *dlm,
                              dlm->name, mres->lockname_len, mres->lockname,
                              from);
                         spin_lock(&res->spinlock);
-                        dlm_lockres_set_refmap_bit(from, res);
+                        dlm_lockres_set_refmap_bit(dlm, res, from);
                         spin_unlock(&res->spinlock);
                         added++;
                         break;
@@ -1965,7 +1972,7 @@ skip_lvb:
                         mlog(0, "%s:%.*s: added lock for node %u, "
                              "setting refmap bit\n", dlm->name,
                              res->lockname.len, res->lockname.name, ml->node);
-                        dlm_lockres_set_refmap_bit(ml->node, res);
+                        dlm_lockres_set_refmap_bit(dlm, res, ml->node);
                         added++;
                 }
                 spin_unlock(&res->spinlock);
@@ -2084,6 +2091,9 @@ static void dlm_finish_local_lockres_recovery(struct dlm_ctxt *dlm,
 
         list_for_each_entry_safe(res, next, &dlm->reco.resources, recovering) {
                 if (res->owner == dead_node) {
+                        mlog(0, "%s: res %.*s, Changing owner from %u to %u\n",
+                             dlm->name, res->lockname.len, res->lockname.name,
+                             res->owner, new_master);
                         list_del_init(&res->recovering);
                         spin_lock(&res->spinlock);
                         /* new_master has our reference from
@@ -2105,40 +2115,30 @@ static void dlm_finish_local_lockres_recovery(struct dlm_ctxt *dlm,
         for (i = 0; i < DLM_HASH_BUCKETS; i++) {
                 bucket = dlm_lockres_hash(dlm, i);
                 hlist_for_each_entry(res, hash_iter, bucket, hash_node) {
-                        if (res->state & DLM_LOCK_RES_RECOVERING) {
-                                if (res->owner == dead_node) {
-                                        mlog(0, "(this=%u) res %.*s owner=%u "
-                                             "was not on recovering list, but "
-                                             "clearing state anyway\n",
-                                             dlm->node_num, res->lockname.len,
-                                             res->lockname.name, new_master);
-                                } else if (res->owner == dlm->node_num) {
-                                        mlog(0, "(this=%u) res %.*s owner=%u "
-                                             "was not on recovering list, "
-                                             "owner is THIS node, clearing\n",
-                                             dlm->node_num, res->lockname.len,
-                                             res->lockname.name, new_master);
-                                } else
-                                        continue;
+                        if (!(res->state & DLM_LOCK_RES_RECOVERING))
+                                continue;
 
-                                if (!list_empty(&res->recovering)) {
-                                        mlog(0, "%s:%.*s: lockres was "
-                                             "marked RECOVERING, owner=%u\n",
-                                             dlm->name, res->lockname.len,
-                                             res->lockname.name, res->owner);
-                                        list_del_init(&res->recovering);
-                                        dlm_lockres_put(res);
-                                }
-                                spin_lock(&res->spinlock);
-                                /* new_master has our reference from
-                                 * the lock state sent during recovery */
-                                dlm_change_lockres_owner(dlm, res, new_master);
-                                res->state &= ~DLM_LOCK_RES_RECOVERING;
-                                if (__dlm_lockres_has_locks(res))
-                                        __dlm_dirty_lockres(dlm, res);
-                                spin_unlock(&res->spinlock);
-                                wake_up(&res->wq);
+                        if (res->owner != dead_node &&
+                            res->owner != dlm->node_num)
+                                continue;
+
+                        if (!list_empty(&res->recovering)) {
+                                list_del_init(&res->recovering);
+                                dlm_lockres_put(res);
                         }
+
+                        /* new_master has our reference from
+                         * the lock state sent during recovery */
+                        mlog(0, "%s: res %.*s, Changing owner from %u to %u\n",
+                             dlm->name, res->lockname.len, res->lockname.name,
+                             res->owner, new_master);
+                        spin_lock(&res->spinlock);
+                        dlm_change_lockres_owner(dlm, res, new_master);
+                        res->state &= ~DLM_LOCK_RES_RECOVERING;
+                        if (__dlm_lockres_has_locks(res))
+                                __dlm_dirty_lockres(dlm, res);
+                        spin_unlock(&res->spinlock);
+                        wake_up(&res->wq);
                 }
         }
 }
@@ -2252,12 +2252,12 @@ static void dlm_free_dead_locks(struct dlm_ctxt *dlm,
                              res->lockname.len, res->lockname.name, freed, dead_node);
                         __dlm_print_one_lock_resource(res);
                 }
-                dlm_lockres_clear_refmap_bit(dead_node, res);
+                dlm_lockres_clear_refmap_bit(dlm, res, dead_node);
         } else if (test_bit(dead_node, res->refmap)) {
                 mlog(0, "%s:%.*s: dead node %u had a ref, but had "
                      "no locks and had not purged before dying\n", dlm->name,
                      res->lockname.len, res->lockname.name, dead_node);
-                dlm_lockres_clear_refmap_bit(dead_node, res);
+                dlm_lockres_clear_refmap_bit(dlm, res, dead_node);
         }
 
         /* do not kick thread yet */
@@ -2324,9 +2324,9 @@ static void dlm_do_local_recovery_cleanup(struct dlm_ctxt *dlm, u8 dead_node)
                         dlm_revalidate_lvb(dlm, res, dead_node);
                         if (res->owner == dead_node) {
                                 if (res->state & DLM_LOCK_RES_DROPPING_REF) {
-                                        mlog(ML_NOTICE, "Ignore %.*s for "
+                                        mlog(ML_NOTICE, "%s: res %.*s, Skip "
                                              "recovery as it is being freed\n",
-                                             res->lockname.len,
+                                             dlm->name, res->lockname.len,
                                              res->lockname.name);
                                 } else
                                         dlm_move_lockres_to_recovery_list(dlm,