6 files changed, 551 insertions, 86 deletions

diff --git a/fs/ocfs2/dlm/dlmcommon.h b/fs/ocfs2/dlm/dlmcommon.h
index 4b6ae2c13b47..b36d0bf77a5a 100644
--- a/fs/ocfs2/dlm/dlmcommon.h
+++ b/fs/ocfs2/dlm/dlmcommon.h
@@ -445,7 +445,9 @@ enum {
         DLM_LOCK_REQUEST_MSG,    /* 515 */
         DLM_RECO_DATA_DONE_MSG,  /* 516 */
         DLM_BEGIN_RECO_MSG,      /* 517 */
-        DLM_FINALIZE_RECO_MSG    /* 518 */
+        DLM_FINALIZE_RECO_MSG,   /* 518 */
+        DLM_QUERY_REGION,        /* 519 */
+        DLM_QUERY_NODEINFO,      /* 520 */
 };
 
 struct dlm_reco_node_data
@@ -727,6 +729,31 @@ struct dlm_cancel_join
         u8 domain[O2NM_MAX_NAME_LEN];
 };
 
+struct dlm_query_region {
+        u8 qr_node;
+        u8 qr_numregions;
+        u8 qr_namelen;
+        u8 pad1;
+        u8 qr_domain[O2NM_MAX_NAME_LEN];
+        u8 qr_regions[O2HB_MAX_REGION_NAME_LEN * O2NM_MAX_REGIONS];
+};
+
+struct dlm_node_info {
+        u8 ni_nodenum;
+        u8 pad1;
+        u16 ni_ipv4_port;
+        u32 ni_ipv4_address;
+};
+
+struct dlm_query_nodeinfo {
+        u8 qn_nodenum;
+        u8 qn_numnodes;
+        u8 qn_namelen;
+        u8 pad1;
+        u8 qn_domain[O2NM_MAX_NAME_LEN];
+        struct dlm_node_info qn_nodes[O2NM_MAX_NODES];
+};
+
 struct dlm_exit_domain
 {
         u8 node_idx;
@@ -1030,6 +1057,7 @@ int dlm_drop_lockres_ref(struct dlm_ctxt *dlm,
                          struct dlm_lock_resource *res);
 void dlm_clean_master_list(struct dlm_ctxt *dlm,
                            u8 dead_node);
+void dlm_force_free_mles(struct dlm_ctxt *dlm);
 int dlm_lock_basts_flushed(struct dlm_ctxt *dlm, struct dlm_lock *lock);
 int __dlm_lockres_has_locks(struct dlm_lock_resource *res);
 int __dlm_lockres_unused(struct dlm_lock_resource *res);
diff --git a/fs/ocfs2/dlm/dlmdebug.c b/fs/ocfs2/dlm/dlmdebug.c
index 5efdd37dfe48..272ec8631a51 100644
--- a/fs/ocfs2/dlm/dlmdebug.c
+++ b/fs/ocfs2/dlm/dlmdebug.c
@@ -493,7 +493,7 @@ static int debug_mle_print(struct dlm_ctxt *dlm, struct debug_buffer *db)
         struct hlist_head *bucket;
         struct hlist_node *list;
         int i, out = 0;
-        unsigned long total = 0, longest = 0, bktcnt;
+        unsigned long total = 0, longest = 0, bucket_count = 0;
 
         out += snprintf(db->buf + out, db->len - out,
                         "Dumping MLEs for Domain: %s\n", dlm->name);
@@ -505,13 +505,13 @@ static int debug_mle_print(struct dlm_ctxt *dlm, struct debug_buffer *db)
                         mle = hlist_entry(list, struct dlm_master_list_entry,
                                           master_hash_node);
                         ++total;
-                        ++bktcnt;
+                        ++bucket_count;
                         if (db->len - out < 200)
                                 continue;
                         out += dump_mle(mle, db->buf + out, db->len - out);
                 }
-                longest = max(longest, bktcnt);
-                bktcnt = 0;
+                longest = max(longest, bucket_count);
+                bucket_count = 0;
         }
         spin_unlock(&dlm->master_lock);
 
@@ -636,8 +636,14 @@ static void *lockres_seq_start(struct seq_file *m, loff_t *pos)
         spin_lock(&dlm->track_lock);
         if (oldres)
                 track_list = &oldres->tracking;
-        else
+        else {
                 track_list = &dlm->tracking_list;
+                if (list_empty(track_list)) {
+                        dl = NULL;
+                        spin_unlock(&dlm->track_lock);
+                        goto bail;
+                }
+        }
 
         list_for_each_entry(res, track_list, tracking) {
                 if (&res->tracking == &dlm->tracking_list)
@@ -660,6 +666,7 @@ static void *lockres_seq_start(struct seq_file *m, loff_t *pos)
         } else
                 dl = NULL;
 
+bail:
         /* passed to seq_show */
         return dl;
 }
@@ -775,7 +782,9 @@ static int debug_state_print(struct dlm_ctxt *dlm, struct debug_buffer *db)
 
         /* Domain: xxxxxxxxxx  Key: 0xdfbac769 */
         out += snprintf(db->buf + out, db->len - out,
-                        "Domain: %s  Key: 0x%08x\n", dlm->name, dlm->key);
+                        "Domain: %s  Key: 0x%08x  Protocol: %d.%d\n",
+                        dlm->name, dlm->key, dlm->dlm_locking_proto.pv_major,
+                        dlm->dlm_locking_proto.pv_minor);
 
         /* Thread Pid: xxx  Node: xxx  State: xxxxx */
         out += snprintf(db->buf + out, db->len - out,
diff --git a/fs/ocfs2/dlm/dlmdomain.c b/fs/ocfs2/dlm/dlmdomain.c
index 153abb5abef0..58a93b953735 100644
--- a/fs/ocfs2/dlm/dlmdomain.c
+++ b/fs/ocfs2/dlm/dlmdomain.c
@@ -128,10 +128,14 @@ static DECLARE_WAIT_QUEUE_HEAD(dlm_domain_events);
  * will have a negotiated version with the same major number and a minor
  * number equal or smaller.  The dlm_ctxt->dlm_locking_proto field should
  * be used to determine what a running domain is actually using.
+ *
+ * New in version 1.1:
+ *      - Message DLM_QUERY_REGION added to support global heartbeat
+ *      - Message DLM_QUERY_NODEINFO added to allow online node removes
  */
 static const struct dlm_protocol_version dlm_protocol = {
         .pv_major = 1,
-        .pv_minor = 0,
+        .pv_minor = 1,
 };
 
 #define DLM_DOMAIN_BACKOFF_MS 200
@@ -142,6 +146,8 @@ static int dlm_assert_joined_handler(struct o2net_msg *msg, u32 len, void *data,
                                      void **ret_data);
 static int dlm_cancel_join_handler(struct o2net_msg *msg, u32 len, void *data,
                                    void **ret_data);
+static int dlm_query_region_handler(struct o2net_msg *msg, u32 len,
+                                    void *data, void **ret_data);
 static int dlm_exit_domain_handler(struct o2net_msg *msg, u32 len, void *data,
                                    void **ret_data);
 static int dlm_protocol_compare(struct dlm_protocol_version *existing,
@@ -693,6 +699,7 @@ void dlm_unregister_domain(struct dlm_ctxt *dlm)
 
                 dlm_mark_domain_leaving(dlm);
                 dlm_leave_domain(dlm);
+                dlm_force_free_mles(dlm);
                 dlm_complete_dlm_shutdown(dlm);
         }
         dlm_put(dlm);
@@ -920,6 +927,370 @@ static int dlm_assert_joined_handler(struct o2net_msg *msg, u32 len, void *data,
         return 0;
 }
 
+static int dlm_match_regions(struct dlm_ctxt *dlm,
+                             struct dlm_query_region *qr)
+{
+        char *local = NULL, *remote = qr->qr_regions;
+        char *l, *r;
+        int localnr, i, j, foundit;
+        int status = 0;
+
+        if (!o2hb_global_heartbeat_active()) {
+                if (qr->qr_numregions) {
+                        mlog(ML_ERROR, "Domain %s: Joining node %d has global "
+                             "heartbeat enabled but local node %d does not\n",
+                             qr->qr_domain, qr->qr_node, dlm->node_num);
+                        status = -EINVAL;
+                }
+                goto bail;
+        }
+
+        if (o2hb_global_heartbeat_active() && !qr->qr_numregions) {
+                mlog(ML_ERROR, "Domain %s: Local node %d has global "
+                     "heartbeat enabled but joining node %d does not\n",
+                     qr->qr_domain, dlm->node_num, qr->qr_node);
+                status = -EINVAL;
+                goto bail;
+        }
+
+        r = remote;
+        for (i = 0; i < qr->qr_numregions; ++i) {
+                mlog(0, "Region %.*s\n", O2HB_MAX_REGION_NAME_LEN, r);
+                r += O2HB_MAX_REGION_NAME_LEN;
+        }
+
+        local = kmalloc(sizeof(qr->qr_regions), GFP_KERNEL);
+        if (!local) {
+                status = -ENOMEM;
+                goto bail;
+        }
+
+        localnr = o2hb_get_all_regions(local, O2NM_MAX_REGIONS);
+
+        /* compare local regions with remote */
+        l = local;
+        for (i = 0; i < localnr; ++i) {
+                foundit = 0;
+                r = remote;
+                for (j = 0; j <= qr->qr_numregions; ++j) {
+                        if (!memcmp(l, r, O2HB_MAX_REGION_NAME_LEN)) {
+                                foundit = 1;
+                                break;
+                        }
+                        r += O2HB_MAX_REGION_NAME_LEN;
+                }
+                if (!foundit) {
+                        status = -EINVAL;
+                        mlog(ML_ERROR, "Domain %s: Region '%.*s' registered "
+                             "in local node %d but not in joining node %d\n",
+                             qr->qr_domain, O2HB_MAX_REGION_NAME_LEN, l,
+                             dlm->node_num, qr->qr_node);
+                        goto bail;
+                }
+                l += O2HB_MAX_REGION_NAME_LEN;
+        }
+
+        /* compare remote with local regions */
+        r = remote;
+        for (i = 0; i < qr->qr_numregions; ++i) {
+                foundit = 0;
+                l = local;
+                for (j = 0; j < localnr; ++j) {
+                        if (!memcmp(r, l, O2HB_MAX_REGION_NAME_LEN)) {
+                                foundit = 1;
+                                break;
+                        }
+                        l += O2HB_MAX_REGION_NAME_LEN;
+                }
+                if (!foundit) {
+                        status = -EINVAL;
+                        mlog(ML_ERROR, "Domain %s: Region '%.*s' registered "
+                             "in joining node %d but not in local node %d\n",
+                             qr->qr_domain, O2HB_MAX_REGION_NAME_LEN, r,
+                             qr->qr_node, dlm->node_num);
+                        goto bail;
+                }
+                r += O2HB_MAX_REGION_NAME_LEN;
+        }
+
+bail:
+        kfree(local);
+
+        return status;
+}
+
+static int dlm_send_regions(struct dlm_ctxt *dlm, unsigned long *node_map)
+{
+        struct dlm_query_region *qr = NULL;
+        int status, ret = 0, i;
+        char *p;
+
+        if (find_next_bit(node_map, O2NM_MAX_NODES, 0) >= O2NM_MAX_NODES)
+                goto bail;
+
+        qr = kzalloc(sizeof(struct dlm_query_region), GFP_KERNEL);
+        if (!qr) {
+                ret = -ENOMEM;
+                mlog_errno(ret);
+                goto bail;
+        }
+
+        qr->qr_node = dlm->node_num;
+        qr->qr_namelen = strlen(dlm->name);
+        memcpy(qr->qr_domain, dlm->name, qr->qr_namelen);
+        /* if local hb, the numregions will be zero */
+        if (o2hb_global_heartbeat_active())
+                qr->qr_numregions = o2hb_get_all_regions(qr->qr_regions,
+                                                         O2NM_MAX_REGIONS);
+
+        p = qr->qr_regions;
+        for (i = 0; i < qr->qr_numregions; ++i, p += O2HB_MAX_REGION_NAME_LEN)
+                mlog(0, "Region %.*s\n", O2HB_MAX_REGION_NAME_LEN, p);
+
+        i = -1;
+        while ((i = find_next_bit(node_map, O2NM_MAX_NODES,
+                                  i + 1)) < O2NM_MAX_NODES) {
+                if (i == dlm->node_num)
+                        continue;
+
+                mlog(0, "Sending regions to node %d\n", i);
+
+                ret = o2net_send_message(DLM_QUERY_REGION, DLM_MOD_KEY, qr,
+                                         sizeof(struct dlm_query_region),
+                                         i, &status);
+                if (ret >= 0)
+                        ret = status;
+                if (ret) {
+                        mlog(ML_ERROR, "Region mismatch %d, node %d\n",
+                             ret, i);
+                        break;
+                }
+        }
+
+bail:
+        kfree(qr);
+        return ret;
+}
+
+static int dlm_query_region_handler(struct o2net_msg *msg, u32 len,
+                                    void *data, void **ret_data)
+{
+        struct dlm_query_region *qr;
+        struct dlm_ctxt *dlm = NULL;
+        int status = 0;
+        int locked = 0;
+
+        qr = (struct dlm_query_region *) msg->buf;
+
+        mlog(0, "Node %u queries hb regions on domain %s\n", qr->qr_node,
+             qr->qr_domain);
+
+        status = -EINVAL;
+
+        spin_lock(&dlm_domain_lock);
+        dlm = __dlm_lookup_domain_full(qr->qr_domain, qr->qr_namelen);
+        if (!dlm) {
+                mlog(ML_ERROR, "Node %d queried hb regions on domain %s "
+                     "before join domain\n", qr->qr_node, qr->qr_domain);
+                goto bail;
+        }
+
+        spin_lock(&dlm->spinlock);
+        locked = 1;
+        if (dlm->joining_node != qr->qr_node) {
+                mlog(ML_ERROR, "Node %d queried hb regions on domain %s "
+                     "but joining node is %d\n", qr->qr_node, qr->qr_domain,
+                     dlm->joining_node);
+                goto bail;
+        }
+
+        /* Support for global heartbeat was added in 1.1 */
+        if (dlm->dlm_locking_proto.pv_major == 1 &&
+            dlm->dlm_locking_proto.pv_minor == 0) {
+                mlog(ML_ERROR, "Node %d queried hb regions on domain %s "
+                     "but active dlm protocol is %d.%d\n", qr->qr_node,
+                     qr->qr_domain, dlm->dlm_locking_proto.pv_major,
+                     dlm->dlm_locking_proto.pv_minor);
+                goto bail;
+        }
+
+        status = dlm_match_regions(dlm, qr);
+
+bail:
+        if (locked)
+                spin_unlock(&dlm->spinlock);
+        spin_unlock(&dlm_domain_lock);
+
+        return status;
+}
+
+static int dlm_match_nodes(struct dlm_ctxt *dlm, struct dlm_query_nodeinfo *qn)
+{
+        struct o2nm_node *local;
+        struct dlm_node_info *remote;
+        int i, j;
+        int status = 0;
+
+        for (j = 0; j < qn->qn_numnodes; ++j)
+                mlog(0, "Node %3d, %pI4:%u\n", qn->qn_nodes[j].ni_nodenum,
+                     &(qn->qn_nodes[j].ni_ipv4_address),
+                     ntohs(qn->qn_nodes[j].ni_ipv4_port));
+
+        for (i = 0; i < O2NM_MAX_NODES && !status; ++i) {
+                local = o2nm_get_node_by_num(i);
+                remote = NULL;
+                for (j = 0; j < qn->qn_numnodes; ++j) {
+                        if (qn->qn_nodes[j].ni_nodenum == i) {
+                                remote = &(qn->qn_nodes[j]);
+                                break;
+                        }
+                }
+
+                if (!local && !remote)
+                        continue;
+
+                if ((local && !remote) || (!local && remote))
+                        status = -EINVAL;
+
+                if (!status &&
+                    ((remote->ni_nodenum != local->nd_num) ||
+                     (remote->ni_ipv4_port != local->nd_ipv4_port) ||
+                     (remote->ni_ipv4_address != local->nd_ipv4_address)))
+                        status = -EINVAL;
+
+                if (status) {
+                        if (remote && !local)
+                                mlog(ML_ERROR, "Domain %s: Node %d (%pI4:%u) "
+                                     "registered in joining node %d but not in "
+                                     "local node %d\n", qn->qn_domain,
+                                     remote->ni_nodenum,
+                                     &(remote->ni_ipv4_address),
+                                     ntohs(remote->ni_ipv4_port),
+                                     qn->qn_nodenum, dlm->node_num);
+                        if (local && !remote)
+                                mlog(ML_ERROR, "Domain %s: Node %d (%pI4:%u) "
+                                     "registered in local node %d but not in "
+                                     "joining node %d\n", qn->qn_domain,
+                                     local->nd_num, &(local->nd_ipv4_address),
+                                     ntohs(local->nd_ipv4_port),
+                                     dlm->node_num, qn->qn_nodenum);
+                        BUG_ON((!local && !remote));
+                }
+
+                if (local)
+                        o2nm_node_put(local);
+        }
+
+        return status;
+}
+
+static int dlm_send_nodeinfo(struct dlm_ctxt *dlm, unsigned long *node_map)
+{
+        struct dlm_query_nodeinfo *qn = NULL;
+        struct o2nm_node *node;
+        int ret = 0, status, count, i;
+
+        if (find_next_bit(node_map, O2NM_MAX_NODES, 0) >= O2NM_MAX_NODES)
+                goto bail;
+
+        qn = kzalloc(sizeof(struct dlm_query_nodeinfo), GFP_KERNEL);
+        if (!qn) {
+                ret = -ENOMEM;
+                mlog_errno(ret);
+                goto bail;
+        }
+
+        for (i = 0, count = 0; i < O2NM_MAX_NODES; ++i) {
+                node = o2nm_get_node_by_num(i);
+                if (!node)
+                        continue;
+                qn->qn_nodes[count].ni_nodenum = node->nd_num;
+                qn->qn_nodes[count].ni_ipv4_port = node->nd_ipv4_port;
+                qn->qn_nodes[count].ni_ipv4_address = node->nd_ipv4_address;
+                mlog(0, "Node %3d, %pI4:%u\n", node->nd_num,
+                     &(node->nd_ipv4_address), ntohs(node->nd_ipv4_port));
+                ++count;
+                o2nm_node_put(node);
+        }
+
+        qn->qn_nodenum = dlm->node_num;
+        qn->qn_numnodes = count;
+        qn->qn_namelen = strlen(dlm->name);
+        memcpy(qn->qn_domain, dlm->name, qn->qn_namelen);
+
+        i = -1;
+        while ((i = find_next_bit(node_map, O2NM_MAX_NODES,
+                                  i + 1)) < O2NM_MAX_NODES) {
+                if (i == dlm->node_num)
+                        continue;
+
+                mlog(0, "Sending nodeinfo to node %d\n", i);
+
+                ret = o2net_send_message(DLM_QUERY_NODEINFO, DLM_MOD_KEY,
+                                         qn, sizeof(struct dlm_query_nodeinfo),
+                                         i, &status);
+                if (ret >= 0)
+                        ret = status;
+                if (ret) {
+                        mlog(ML_ERROR, "node mismatch %d, node %d\n", ret, i);
+                        break;
+                }
+        }
+
+bail:
+        kfree(qn);
+        return ret;
+}
+
+static int dlm_query_nodeinfo_handler(struct o2net_msg *msg, u32 len,
+                                      void *data, void **ret_data)
+{
+        struct dlm_query_nodeinfo *qn;
+        struct dlm_ctxt *dlm = NULL;
+        int locked = 0, status = -EINVAL;
+
+        qn = (struct dlm_query_nodeinfo *) msg->buf;
+
+        mlog(0, "Node %u queries nodes on domain %s\n", qn->qn_nodenum,
+             qn->qn_domain);
+
+        spin_lock(&dlm_domain_lock);
+        dlm = __dlm_lookup_domain_full(qn->qn_domain, qn->qn_namelen);
+        if (!dlm) {
+                mlog(ML_ERROR, "Node %d queried nodes on domain %s before "
+                     "join domain\n", qn->qn_nodenum, qn->qn_domain);
+                goto bail;
+        }
+
+        spin_lock(&dlm->spinlock);
+        locked = 1;
+        if (dlm->joining_node != qn->qn_nodenum) {
+                mlog(ML_ERROR, "Node %d queried nodes on domain %s but "
+                     "joining node is %d\n", qn->qn_nodenum, qn->qn_domain,
+                     dlm->joining_node);
+                goto bail;
+        }
+
+        /* Support for node query was added in 1.1 */
+        if (dlm->dlm_locking_proto.pv_major == 1 &&
+            dlm->dlm_locking_proto.pv_minor == 0) {
+                mlog(ML_ERROR, "Node %d queried nodes on domain %s "
+                     "but active dlm protocol is %d.%d\n", qn->qn_nodenum,
+                     qn->qn_domain, dlm->dlm_locking_proto.pv_major,
+                     dlm->dlm_locking_proto.pv_minor);
+                goto bail;
+        }
+
+        status = dlm_match_nodes(dlm, qn);
+
+bail:
+        if (locked)
+                spin_unlock(&dlm->spinlock);
+        spin_unlock(&dlm_domain_lock);
+
+        return status;
+}
+
 static int dlm_cancel_join_handler(struct o2net_msg *msg, u32 len, void *data,
                                    void **ret_data)
 {
@@ -1240,6 +1611,20 @@ static int dlm_try_to_join_domain(struct dlm_ctxt *dlm)
         set_bit(dlm->node_num, dlm->domain_map);
         spin_unlock(&dlm->spinlock);
 
+        /* Support for global heartbeat and node info was added in 1.1 */
+        if (dlm_protocol.pv_major > 1 || dlm_protocol.pv_minor > 0) {
+                status = dlm_send_nodeinfo(dlm, ctxt->yes_resp_map);
+                if (status) {
+                        mlog_errno(status);
+                        goto bail;
+                }
+                status = dlm_send_regions(dlm, ctxt->yes_resp_map);
+                if (status) {
+                        mlog_errno(status);
+                        goto bail;
+                }
+        }
+
         dlm_send_join_asserts(dlm, ctxt->yes_resp_map);
 
         /* Joined state *must* be set before the joining node
@@ -1806,7 +2191,21 @@ static int dlm_register_net_handlers(void)
                                         sizeof(struct dlm_cancel_join),
                                         dlm_cancel_join_handler,
                                         NULL, NULL, &dlm_join_handlers);
+        if (status)
+                goto bail;
+
+        status = o2net_register_handler(DLM_QUERY_REGION, DLM_MOD_KEY,
+                                        sizeof(struct dlm_query_region),
+                                        dlm_query_region_handler,
+                                        NULL, NULL, &dlm_join_handlers);
 
+        if (status)
+                goto bail;
+
+        status = o2net_register_handler(DLM_QUERY_NODEINFO, DLM_MOD_KEY,
+                                        sizeof(struct dlm_query_nodeinfo),
+                                        dlm_query_nodeinfo_handler,
+                                        NULL, NULL, &dlm_join_handlers);
 bail:
         if (status < 0)
                 dlm_unregister_net_handlers();
diff --git a/fs/ocfs2/dlm/dlmmaster.c b/fs/ocfs2/dlm/dlmmaster.c
index 94b97fc6a88e..f564b0e5f80d 100644
--- a/fs/ocfs2/dlm/dlmmaster.c
+++ b/fs/ocfs2/dlm/dlmmaster.c
@@ -511,8 +511,6 @@ static void dlm_lockres_release(struct kref *kref)
 
         atomic_dec(&dlm->res_cur_count);
 
-        dlm_put(dlm);
-
         if (!hlist_unhashed(&res->hash_node) ||
             !list_empty(&res->granted) ||
             !list_empty(&res->converting) ||
@@ -585,8 +583,6 @@ static void dlm_init_lockres(struct dlm_ctxt *dlm,
         res->migration_pending = 0;
         res->inflight_locks = 0;
 
-        /* put in dlm_lockres_release */
-        dlm_grab(dlm);
         res->dlm = dlm;
 
         kref_init(&res->refs);
@@ -3050,8 +3046,6 @@ int dlm_migrate_request_handler(struct o2net_msg *msg, u32 len, void *data,
         /* check for pre-existing lock */
         spin_lock(&dlm->spinlock);
         res = __dlm_lookup_lockres(dlm, name, namelen, hash);
-        spin_lock(&dlm->master_lock);
-
         if (res) {
                 spin_lock(&res->spinlock);
                 if (res->state & DLM_LOCK_RES_RECOVERING) {
@@ -3069,14 +3063,15 @@ int dlm_migrate_request_handler(struct o2net_msg *msg, u32 len, void *data,
                 spin_unlock(&res->spinlock);
         }
 
+        spin_lock(&dlm->master_lock);
         /* ignore status.  only nonzero status would BUG. */
         ret = dlm_add_migration_mle(dlm, res, mle, &oldmle,
                                     name, namelen,
                                     migrate->new_master,
                                     migrate->master);
 
-unlock:
         spin_unlock(&dlm->master_lock);
+unlock:
         spin_unlock(&dlm->spinlock);
 
         if (oldmle) {
@@ -3438,3 +3433,43 @@ void dlm_lockres_release_ast(struct dlm_ctxt *dlm,
         wake_up(&res->wq);
         wake_up(&dlm->migration_wq);
 }
+
+void dlm_force_free_mles(struct dlm_ctxt *dlm)
+{
+        int i;
+        struct hlist_head *bucket;
+        struct dlm_master_list_entry *mle;
+        struct hlist_node *tmp, *list;
+
+        /*
+         * We notified all other nodes that we are exiting the domain and
+         * marked the dlm state to DLM_CTXT_LEAVING. If any mles are still
+         * around we force free them and wake any processes that are waiting
+         * on the mles
+         */
+        spin_lock(&dlm->spinlock);
+        spin_lock(&dlm->master_lock);
+
+        BUG_ON(dlm->dlm_state != DLM_CTXT_LEAVING);
+        BUG_ON((find_next_bit(dlm->domain_map, O2NM_MAX_NODES, 0) < O2NM_MAX_NODES));
+
+        for (i = 0; i < DLM_HASH_BUCKETS; i++) {
+                bucket = dlm_master_hash(dlm, i);
+                hlist_for_each_safe(list, tmp, bucket) {
+                        mle = hlist_entry(list, struct dlm_master_list_entry,
+                                          master_hash_node);
+                        if (mle->type != DLM_MLE_BLOCK) {
+                                mlog(ML_ERROR, "bad mle: %p\n", mle);
+                                dlm_print_one_mle(mle);
+                        }
+                        atomic_set(&mle->woken, 1);
+                        wake_up(&mle->wq);
+
+                        __dlm_unlink_mle(dlm, mle);
+                        __dlm_mle_detach_hb_events(dlm, mle);
+                        __dlm_put_mle(mle);
+                }
+        }
+        spin_unlock(&dlm->master_lock);
+        spin_unlock(&dlm->spinlock);
+}
diff --git a/fs/ocfs2/dlm/dlmrecovery.c b/fs/ocfs2/dlm/dlmrecovery.c
index 9dfaac73b36d..aaaffbcbe916 100644
--- a/fs/ocfs2/dlm/dlmrecovery.c
+++ b/fs/ocfs2/dlm/dlmrecovery.c
@@ -1997,6 +1997,8 @@ void dlm_move_lockres_to_recovery_list(struct dlm_ctxt *dlm,
         struct list_head *queue;
         struct dlm_lock *lock, *next;
 
+        assert_spin_locked(&dlm->spinlock);
+        assert_spin_locked(&res->spinlock);
         res->state |= DLM_LOCK_RES_RECOVERING;
         if (!list_empty(&res->recovering)) {
                 mlog(0,
@@ -2326,19 +2328,15 @@ static void dlm_do_local_recovery_cleanup(struct dlm_ctxt *dlm, u8 dead_node)
                         /* zero the lvb if necessary */
                         dlm_revalidate_lvb(dlm, res, dead_node);
                         if (res->owner == dead_node) {
-                                if (res->state & DLM_LOCK_RES_DROPPING_REF)
-                                        mlog(0, "%s:%.*s: owned by "
-                                             "dead node %u, this node was "
-                                             "dropping its ref when it died. "
-                                             "continue, dropping the flag.\n",
-                                             dlm->name, res->lockname.len,
-                                             res->lockname.name, dead_node);
-
-                                /* the wake_up for this will happen when the
-                                 * RECOVERING flag is dropped later */
-                                res->state &= ~DLM_LOCK_RES_DROPPING_REF;
+                                if (res->state & DLM_LOCK_RES_DROPPING_REF) {
+                                        mlog(ML_NOTICE, "Ignore %.*s for "
+                                             "recovery as it is being freed\n",
+                                             res->lockname.len,
+                                             res->lockname.name);
+                                } else
+                                        dlm_move_lockres_to_recovery_list(dlm,
+                                                                          res);
 
-                                dlm_move_lockres_to_recovery_list(dlm, res);
                         } else if (res->owner == dlm->node_num) {
                                 dlm_free_dead_locks(dlm, res, dead_node);
                                 __dlm_lockres_calc_usage(dlm, res);
diff --git a/fs/ocfs2/dlm/dlmthread.c b/fs/ocfs2/dlm/dlmthread.c
index d4f73ca68fe5..2211acf33d9b 100644
--- a/fs/ocfs2/dlm/dlmthread.c
+++ b/fs/ocfs2/dlm/dlmthread.c
@@ -92,19 +92,27 @@ int __dlm_lockres_has_locks(struct dlm_lock_resource *res)
  * truly ready to be freed. */
 int __dlm_lockres_unused(struct dlm_lock_resource *res)
 {
-        if (!__dlm_lockres_has_locks(res) &&
-            (list_empty(&res->dirty) && !(res->state & DLM_LOCK_RES_DIRTY))) {
-                /* try not to scan the bitmap unless the first two
-                 * conditions are already true */
-                int bit = find_next_bit(res->refmap, O2NM_MAX_NODES, 0);
-                if (bit >= O2NM_MAX_NODES) {
-                        /* since the bit for dlm->node_num is not
-                         * set, inflight_locks better be zero */
-                        BUG_ON(res->inflight_locks != 0);
-                        return 1;
-                }
-        }
-        return 0;
+        int bit;
+
+        if (__dlm_lockres_has_locks(res))
+                return 0;
+
+        if (!list_empty(&res->dirty) || res->state & DLM_LOCK_RES_DIRTY)
+                return 0;
+
+        if (res->state & DLM_LOCK_RES_RECOVERING)
+                return 0;
+
+        bit = find_next_bit(res->refmap, O2NM_MAX_NODES, 0);
+        if (bit < O2NM_MAX_NODES)
+                return 0;
+
+        /*
+         * since the bit for dlm->node_num is not set, inflight_locks better
+         * be zero
+         */
+        BUG_ON(res->inflight_locks != 0);
+        return 1;
 }
 
 
@@ -152,45 +160,25 @@ void dlm_lockres_calc_usage(struct dlm_ctxt *dlm,
         spin_unlock(&dlm->spinlock);
 }
 
-static int dlm_purge_lockres(struct dlm_ctxt *dlm,
+static void dlm_purge_lockres(struct dlm_ctxt *dlm,
                              struct dlm_lock_resource *res)
 {
         int master;
         int ret = 0;
 
-        spin_lock(&res->spinlock);
-        if (!__dlm_lockres_unused(res)) {
-                mlog(0, "%s:%.*s: tried to purge but not unused\n",
-                     dlm->name, res->lockname.len, res->lockname.name);
-                __dlm_print_one_lock_resource(res);
-                spin_unlock(&res->spinlock);
-                BUG();
-        }
-
-        if (res->state & DLM_LOCK_RES_MIGRATING) {
-                mlog(0, "%s:%.*s: Delay dropref as this lockres is "
-                     "being remastered\n", dlm->name, res->lockname.len,
-                     res->lockname.name);
-                /* Re-add the lockres to the end of the purge list */
-                if (!list_empty(&res->purge)) {
-                        list_del_init(&res->purge);
-                        list_add_tail(&res->purge, &dlm->purge_list);
-                }
-                spin_unlock(&res->spinlock);
-                return 0;
-        }
+        assert_spin_locked(&dlm->spinlock);
+        assert_spin_locked(&res->spinlock);
 
         master = (res->owner == dlm->node_num);
 
-        if (!master)
-                res->state |= DLM_LOCK_RES_DROPPING_REF;
-        spin_unlock(&res->spinlock);
 
         mlog(0, "purging lockres %.*s, master = %d\n", res->lockname.len,
              res->lockname.name, master);
 
         if (!master) {
+                res->state |= DLM_LOCK_RES_DROPPING_REF;
                 /* drop spinlock...  retake below */
+                spin_unlock(&res->spinlock);
                 spin_unlock(&dlm->spinlock);
 
                 spin_lock(&res->spinlock);
@@ -208,31 +196,35 @@ static int dlm_purge_lockres(struct dlm_ctxt *dlm,
                 mlog(0, "%s:%.*s: dlm_deref_lockres returned %d\n",
                      dlm->name, res->lockname.len, res->lockname.name, ret);
                 spin_lock(&dlm->spinlock);
+                spin_lock(&res->spinlock);
         }
 
-        spin_lock(&res->spinlock);
         if (!list_empty(&res->purge)) {
                 mlog(0, "removing lockres %.*s:%p from purgelist, "
                      "master = %d\n", res->lockname.len, res->lockname.name,
                      res, master);
                 list_del_init(&res->purge);
-                spin_unlock(&res->spinlock);
                 dlm_lockres_put(res);
                 dlm->purge_count--;
-        } else
-                spin_unlock(&res->spinlock);
+        }
+
+        if (!__dlm_lockres_unused(res)) {
+                mlog(ML_ERROR, "found lockres %s:%.*s: in use after deref\n",
+                     dlm->name, res->lockname.len, res->lockname.name);
+                __dlm_print_one_lock_resource(res);
+                BUG();
+        }
 
         __dlm_unhash_lockres(res);
 
         /* lockres is not in the hash now.  drop the flag and wake up
          * any processes waiting in dlm_get_lock_resource. */
         if (!master) {
-                spin_lock(&res->spinlock);
                 res->state &= ~DLM_LOCK_RES_DROPPING_REF;
                 spin_unlock(&res->spinlock);
                 wake_up(&res->wq);
-        }
-        return 0;
+        } else
+                spin_unlock(&res->spinlock);
 }
 
 static void dlm_run_purge_list(struct dlm_ctxt *dlm,
@@ -251,17 +243,7 @@ static void dlm_run_purge_list(struct dlm_ctxt *dlm,
                 lockres = list_entry(dlm->purge_list.next,
                                      struct dlm_lock_resource, purge);
 
-                /* Status of the lockres *might* change so double
-                 * check. If the lockres is unused, holding the dlm
-                 * spinlock will prevent people from getting and more
-                 * refs on it -- there's no need to keep the lockres
-                 * spinlock. */
                 spin_lock(&lockres->spinlock);
-                unused = __dlm_lockres_unused(lockres);
-                spin_unlock(&lockres->spinlock);
-
-                if (!unused)
-                        continue;
 
                 purge_jiffies = lockres->last_used +
                         msecs_to_jiffies(DLM_PURGE_INTERVAL_MS);
@@ -273,15 +255,29 @@ static void dlm_run_purge_list(struct dlm_ctxt *dlm,
                          * in tail order, we can stop at the first
                          * unpurgable resource -- anyone added after
                          * him will have a greater last_used value */
+                        spin_unlock(&lockres->spinlock);
                         break;
                 }
 
+                /* Status of the lockres *might* change so double
+                 * check. If the lockres is unused, holding the dlm
+                 * spinlock will prevent people from getting and more
+                 * refs on it. */
+                unused = __dlm_lockres_unused(lockres);
+                if (!unused ||
+                    (lockres->state & DLM_LOCK_RES_MIGRATING)) {
+                        mlog(0, "lockres %s:%.*s: is in use or "
+                             "being remastered, used %d, state %d\n",
+                             dlm->name, lockres->lockname.len,
+                             lockres->lockname.name, !unused, lockres->state);
+                        list_move_tail(&dlm->purge_list, &lockres->purge);
+                        spin_unlock(&lockres->spinlock);
+                        continue;
+                }
+
                 dlm_lockres_get(lockres);
 
-                /* This may drop and reacquire the dlm spinlock if it
-                 * has to do migration. */
-                if (dlm_purge_lockres(dlm, lockres))
-                        BUG();
+                dlm_purge_lockres(dlm, lockres);
 
                 dlm_lockres_put(lockres);